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

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

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

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

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

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

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

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    database_server: str
    database_name: str
    pool_size: int = 20
    pool_idle_timeout: int = 300
    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 автоматически считывает имена переменных среды с заглавной буквы.

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()) as conn:
        with conn.cursor() as cursor:
            yield cursor

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

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

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

Добавьте обработчики после 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 в развернутых средах.

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

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

Last updated on 2026-09-23