Тестирование и развертывание приложений FastAPI на mssql-python

После создания приложения FastAPI на mssql-python настройте его для развертывания, повторного использования соединений, обработки ошибок, аутентификации и автоматизированного тестирования.

Необходимые условия

Выполните инструкции Использование mssql-python с FastAPI или используйте эквивалентное приложение FastAPI, работающее с образцом базы данных AdventureWorksLT. Зависимость аутентификации в этой статье обращается к SalesLT.Customer.
Установите рабочие и тестовые зависимости:
```
pip install pydantic-settings pyjwt pytest httpx2
```

Настройка параметров развертывания

Используйте настройки Pydantic для загрузки значений, специфичных для развертывания, из переменных среды. Этот подход не допускает секретов исходного кода и предоставляет каждой среде собственную конфигурацию базы данных, пула и аутентификации.

В корне проекта создайте config.py рядом database.py:

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    database_server: str
    database_name: str
    pool_size: int = 20
    pool_idle_timeout: int = 300
    login_timeout: int = 30
    jwt_secret: str


settings = Settings()


def get_connection_string() -> str:
    return (
        f"Server={settings.database_server};"
        f"Database={settings.database_name};"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes"
    )

Установите DATABASE_SERVER, DATABASE_NAME, и JWT_SECRET в среде развертывания. Pydantic Settings автоматически считывает имена переменных среды с заглавной буквы.

Для локального тестирования замените заполнятели базы данных и установите переменные в активированной среде. Запускайте pytest позже с того же терминала.

Windows
Linux или macOS

$env:DATABASE_SERVER = "<server>.database.windows.net"
$env:DATABASE_NAME = "<database>"
$env:JWT_SECRET = python -c "import secrets; print(secrets.token_urlsafe(32))"

export DATABASE_SERVER="<server>.database.windows.net"
export DATABASE_NAME="<database>"
export JWT_SECRET="$(python -c 'import secrets; print(secrets.token_urlsafe(32))')"

Note

ActiveDirectoryDefault последовательно проверяет несколько поставщиков учетных данных. В продакшене укажите режим аутентификации для развернутой идентичности, например ActiveDirectoryMSI , для управляемой идентичности, чтобы избежать прохождения по цепочке учетных данных. Для доступных режимов см. Аутентификация Microsoft Entra с mssql-python.

Настройка пула подключений

MSSQL-Python по умолчанию поддерживает пул соединений. Настройте пул один раз, до того, как приложение создаст своё первое соединение. Определите размер пула для ожидаемой параллельной работы с базой данных и уровня сервиса базы данных.

Обновите database.py, чтобы использовать параметры развертывания:

from collections.abc import Generator

import mssql_python

from config import get_connection_string, settings


mssql_python.pooling(
    max_size=settings.pool_size,
    idle_timeout=settings.pool_idle_timeout,
)


def get_db_dependency() -> Generator:
    with mssql_python.connect(
        get_connection_string(), timeout=settings.login_timeout
    ) as conn:
        with conn.cursor() as cursor:
            yield cursor

Менеджер контекста соединения коммитирует после успешной обработки запроса, откатывается при возникновении исключения при обработке запроса, и закрывает соединение. Закрытие соединения возвращает его в пул. Для ключей пула, размера, изоляции идентичности и рекомендаций по исчерпанию см. раздел «Пул соединений с mssql-python».

С провайдером по умолчанию msodbcsql18LOGIN_TIMEOUT устанавливает тайм-аут входа в систему и предоставляет временный нативный тайм-аут операции соединения во время валидации соединения из пула, в секундах. Если проверка не проходит, драйвер разрывает соединение и пытается установить новое соединение. Многократные операции валидации, очистка и замена могут занять больше времени, чем настроенное значение. Эта настройка не является общим тайм-аутом проверки соединения или крайним сроком выполнения запроса и не устанавливает тайм-аут запроса.

С провайдером opt-in mssql-odbc, не гарантируется, что сброс соединения из пула будет ограничен тайм-аутом входа. Для получения дополнительной информации см. раздел «Ограничения поставщика».

Обработка ошибок базы данных

Регистрируйте обработчики исключений так, чтобы сбои базы данных возвращали последовательные ответы без раскрытия деталей соединения, запросов или текста ошибок сервера.

Добавьте обработчики после app = FastAPI(...) в main.py:

import mssql_python
from fastapi import Request
from fastapi.responses import JSONResponse


@app.exception_handler(mssql_python.IntegrityError)
async def integrity_exception_handler(
    request: Request,
    exc: mssql_python.IntegrityError,
):
    return JSONResponse(
        status_code=409,
        content={
            "detail": "The request conflicts with existing data.",
            "type": "integrity_error",
        },
    )


@app.exception_handler(mssql_python.DatabaseError)
async def database_exception_handler(
    request: Request,
    exc: mssql_python.DatabaseError,
):
    return JSONResponse(
        status_code=500,
        content={
            "detail": "A database operation failed.",
            "type": "database_error",
        },
    )

Зарегистрируйте исключение через защищённый телеметрический конвейер вашего приложения перед тем, как возвращать ответ. Для иерархии исключений и обработки SQLSTATE см. Обработка ошибок и коды SQLSTATE для mssql-python.

Добавить зависимости аутентификации

Свяжите зависимости FastAPI в цепочку, чтобы проверить JSON Web Token (JWT), загрузить соответствующего клиента AdventureWorksLT и сделать этого клиента доступным для защищённых маршрутов. Проверьте токен до получения соединения с базой данных, чтобы недействительный токен не использовал пуловое соединение.

Создайте auth.py:

import jwt
from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

from config import settings
from database import get_db_dependency


security = HTTPBearer()


def get_customer_id(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> int:
    try:
        payload = jwt.decode(
            credentials.credentials,
            settings.jwt_secret,
            algorithms=["HS256"],
        )
        customer_id = int(payload["sub"])
    except (KeyError, TypeError, ValueError):
        raise HTTPException(status_code=401, detail="Invalid token subject")
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

    return customer_id


def get_current_customer(
    customer_id: int = Depends(get_customer_id),
    cursor = Depends(get_db_dependency),
):
    cursor.execute(
        """
        SELECT CustomerID, FirstName, LastName
        FROM SalesLT.Customer
        WHERE CustomerID = %(id)s
        """,
        {"id": customer_id},
    )
    customer = cursor.fetchone()
    if customer is None:
        raise HTTPException(status_code=401, detail="Customer not found")

    return {
        "id": customer.CustomerID,
        "first_name": customer.FirstName,
        "last_name": customer.LastName,
    }

Импортируйте зависимость и добавьте защищённый маршрут в main.py:

from auth import get_current_customer


@app.get("/me")
def get_me(current_customer: dict = Depends(get_current_customer)):
    return current_customer

Используйте провайдера идентификации для выдачи и ротации ключей подписи. Для HS256 задайте JWT_SECRET как случайное значение длиной не менее 32 байт. Не храните секрет подписи производства в репозитории или в изображении.

Тестирование приложения

FastAPI TestClient отправляет запросы приложению без запуска HTTP-сервера. Следующие интеграционные тесты используют настроенную базу данных.

Создайте test_api.py:

import uuid

from fastapi.testclient import TestClient

from main import app


client = TestClient(app)


def test_list_products():
    response = client.get("/products")
    assert response.status_code == 200
    data = response.json()
    assert "items" in data
    assert "total" in data


def test_create_product():
    suffix = uuid.uuid4().hex[:8]
    response = client.post(
        "/products",
        json={
            "name": f"Test Product {suffix}",
            "product_number": f"TEST-{suffix}",
            "price": 19.99,
            "color": "Red",
            "size": "M",
            "category_id": 1,
        },
    )
    assert response.status_code == 201
    data = response.json()
    assert data["product_number"] == f"TEST-{suffix}"
    assert data["price"] == 19.99


def test_get_product_not_found():
    response = client.get("/products/99999")
    assert response.status_code == 404


def test_health_check():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"

Проведите тесты с корня проекта:

pytest

Эти тесты используют настроенную базу данных, а test_create_product вставляют строку в SalesLT.Product. Используйте выделенную тестовую базу данных и сбрасывайте данные в ней между тестовыми запусками.

Контрольный список развертывания

Установите DATABASE_SERVER, DATABASE_NAME, и JWT_SECRET через секретные и конфигурационные хранилища платформы.
Используйте выделенный идентификатор Microsoft Entra с минимальными необходимыми правами доступа к базе данных.
Установите размер пула ниже лимита соединения базы данных и оставьте пропускную способность для административного доступа и других рабочих нагрузок.
Запускайте интеграционные тесты базы данных с изолированной тестовой базой данных.
Настройте защищённую телеметрию для исключений в базе данных, задержки запросов и исчерпания пула.
Запускайте Uvicorn без --reload в развернутых средах.

Обратная связь

Были ли сведения на этой странице полезными?

Последнее обновление: 2026-11-05