Testujte a nasazujte FastAPI aplikace pomocí mssql-python

Poté, co postavíte FastAPI aplikaci pomocí mssql-python, nakonfigurujte ji pro nasazení, opětovné použití připojení, zpracování chyb, autentizaci a automatizované testování.

Předpoklady

• Kompletně používejte mssql-python s FastAPI, nebo mějte ekvivalentní aplikaci FastAPI, která využívá ukázkovou databázi AdventureWorksLT. Závislost ověřování v tomto článku se dotazuje na SalesLT.Customer.

• Nainstalujte produkční a testovací závislosti:

  pip install pydantic-settings pyjwt pytest httpx
  

Konfigurace nastavení nasazení

Použijte Pydantic Settings k načtení hodnot specifických pro nasazení z proměnných prostředí. Tento přístup udržuje tajemství mimo zdrojový kód a každému prostředí poskytuje vlastní databázi, pool a konfiguraci autentizace.

Vytvořit 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"
    )

Nastavte v prostředí nasazení DATABASE_SERVER, DATABASE_NAME a JWT_SECRET. Pydantic Settings automaticky čte názvy proměnných prostředí velkými písmeny.

Note

ActiveDirectoryDefault zkouší více poskytovatelů přihlašovacích údajů za sebou. V produkci specifikujte autentizační režim pro nasazenou identitu, například ActiveDirectoryMSI pro spravovanou identitu, abyste se vyhnuli procházení řetězce přihlašovacích údajů. Pro dostupné režimy viz Ověřování Microsoft Entra pomocí mssql-python.

Konfigurace sdružování připojení

mssql-python ve výchozím nastavení umožňuje poolování spojení. Konfigurujte pool jednou, než aplikace vytvoří první připojení. Nastavte velikost fondu připojení podle očekávaného souběžného zatížení databáze aplikací a úrovně databázové služby.

Aktualizace database.py pro použití nastavení nasazení:

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

Správce kontextu připojení po úspěšném zpracování požadavku potvrdí transakci, pokud zpracování požadavku vyvolá výjimku, provede rollback a uzavře spojení. Po uzavření spojení se vrací do poolu. Klíče fondu připojení, dimenzování, izolaci identit a doporučení ohledně vyčerpání najdete v tématu Connection pooling with mssql-python.

Řešení chyb v databázi

Zaregistrujte obslužné nástroje pro výjimky, aby selhání databáze vracela konzistentní odpovědi bez zveřejňování detailů spojení, dotazů nebo chybového textu serveru.

Přidejte obslužné rutiny za app = FastAPI(...) v 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",
        },
    )

Před vrácením odpovědi zaznamenejte výjimku prostřednictvím chráněného kanálu telemetrie vaší aplikace. Pro hierarchii výjimek a zpracování SQLSTATE viz Zpracování chyb a kódy SQLSTATE pro mssql-python.

Přidat autentizační závislosti

Propojte závislosti FastAPI tak, aby ověřily token JWT (JSON Web Token), načetly odpovídajícího zákazníka AdventureWorksLT a zpřístupnily tohoto zákazníka chráněným trasám. Ověřte token před navázáním připojení k databázi, aby neplatný token nevyužil připojení z poolu.

Vytvořit 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,
    }

Importujte závislost a přidejte chráněnou trasu do main.py:

from auth import get_current_customer


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

Používejte poskytovatele identity k vydávání a rotaci podpisových klíčů. Pro HS256 nastavte JWT_SECRET na alespoň 32 náhodných bajtů. Neukládejte tajný klíč pro produkční podepisování do repozitáře ani do image.

Testování aplikace

TestClient FastAPI odesílá požadavky aplikaci, aniž by spouštěl HTTP server. Následující integrační testy využívají nakonfigurovanou databázi.

Vytvořit 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"

Spusť testy z kořene projektu:

pytest

Tyto testy využívají nakonfigurovanou databázi a test_create_product vkládají řádek do SalesLT.Product. Používejte dedikovanou testovací databázi a mezi testy resetujte její data.

Kontrolní seznam nasazení

• Nastavte DATABASE_SERVER, DATABASE_NAME a JWT_SECRET prostřednictvím úložišť tajných klíčů a konfigurace platformy pro nasazení.
• Použijte dedikovanou identitu Microsoft Entra s minimálními požadovanými oprávněními k databázi.
• Nastavte velikost poolu pod limit připojení databáze a nechte kapacitu pro administrátorský přístup a další pracovní zátěže.
• Spusť testy integrace databáze proti izolované testovací databázi.
• Nakonfigurujte chráněnou telemetrii pro výjimky databáze, latenci požadavků a vyčerpání poolu.
• Spusťte Uvicorn bez --reload v nasazených prostředích.

Související obsah

• Použijte mssql-python s FastAPI
• Sdružování připojení v mssql-python
• Zpracování chyb a SQLSTATE kódy pro mssql-python
• Ověřování Microsoft Entra pomocí mssql-python
• Osvědčené postupy bezpečnosti pro aplikace mssql-python