FastAPI-Anwendungen mit mssql-python testen und bereitstellen

Nachdem du eine FastAPI-Anwendung mit mssql-python gebaut hast, konfiguriere sie für Deployment, Verbindungswiederverwendung, Fehlerbehandlung, Authentifizierung und automatisiertes Testen.

Prerequisites

  • Komplett Verwenden Sie mssql-python mit FastAPI oder haben Sie eine entsprechende FastAPI-Anwendung, die die AdventureWorksLT-Beispieldatenbank verwendet. Die in diesem Artikel verwendete Authentifizierungsabhängigkeit fragt SalesLT.Customer ab.

  • Installieren Sie die Produktions- und Testabhängigkeiten:

    pip install pydantic-settings pyjwt pytest httpx

Bereitstellungseinstellungen konfigurieren

Verwenden Sie Pydantic Settings, um deployment-spezifische Werte aus Umgebungsvariablen zu laden. Dieser Ansatz hält Geheimnisse aus dem Quellcode heraus und gibt jeder Umgebung eine eigene Datenbank-, Pool- und Authentifizierungskonfiguration.

Erstellen Sie 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"
    )

Setze DATABASE_SERVER, DATABASE_NAME, und JWT_SECRET in der Deployment-Umgebung. Pydantic Settings liest automatisch die Namen der Großbuchstaben-Umgebungsvariablen aus.

Note

ActiveDirectoryDefault probiert mehrere Anmeldeinformationsanbieter der Reihe nach aus. In der Produktion spezifizieren Sie den Authentifizierungsmodus für die bereitgestellte Identität, zum Beispiel ActiveDirectoryMSI für eine verwaltete Identität, um ein Durchlaufen der Zugangsdatenkette zu vermeiden. Für verfügbare Modi siehe Microsoft Entra-Authentifizierung mit mssql-python.

Konfigurieren von Verbindungspooling

MSSQL-Python ermöglicht standardmäßig Connection Pooling. Konfigurieren Sie den Pool einmal, bevor die Anwendung ihre erste Verbindung erstellt. Dimensionieren Sie den Pool für die erwartete gleichzeitige Datenbankarbeit der Anwendung und die Datenbankservice-Ebene.

Aktualisieren Sie database.py, um die Bereitstellungseinstellungen zu verwenden:

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

Der Verbindungs-Kontextmanager commitet nach erfolgreicher Anfrageverarbeitung, rollt zurück, wenn die Anfrageverarbeitung eine Ausnahme auslöst, und schließt die Verbindung. Das Schließen der Verbindung bringt es zurück in den Pool. Für Poolschlüssel, Größen, Identitätsisolation und Exhaustion-Anleitung siehe Connection Pooling mit mssql-python.

Datenbankfehler behandeln

Registrieren Sie Ausnahmehandler, damit Datenbankfehler konsistente Antworten zurückgeben, ohne Verbindungsdetails, Abfragen oder Serverfehler anzuzeigen.

Fügen Sie die Handler nach app = FastAPI(...) in main.py hinzu:

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",
        },
    )

Loggen Sie die Ausnahme durch die geschützte Telemetriepipeline Ihrer Anwendung, bevor Sie die Antwort zurückgeben. Für die Ausnahmehierarchie und SQLSTATE-Behandlung siehe Error handling und SQLSTATE-Codes für mssql-python.

Authentifizierungsabhängigkeiten hinzufügen

Verketten Sie FastAPI-Abhängigkeiten, um einen JSON Web Token (JWT) zu validieren, den zugehörigen AdventureWorksLT-Kunden zu laden und diesen Kunden für geschützte Routen verfügbar zu machen. Validiere das Token, bevor du eine Datenbankverbindung erwirbst, damit ein ungültiger Token keine Pool-Verbindung verwendet.

Erstellen Sie 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,
    }

Importiere die Abhängigkeit und füge eine geschützte Route hinzu:main.py

from auth import get_current_customer


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

Nutzen Sie einen Identitätsanbieter, um Unterzeichnungsschlüssel auszustellen und zu rotieren. Für HS256 setzen JWT_SECRET Sie auf mindestens 32 zufällige Bytes. Speichere keinen Signaturschlüssel für die Produktion im Repository oder in einem Image.

Testen der App

FastAPI sendet TestClient Anfragen an die Anwendung, ohne einen HTTP-Server zu starten. Die folgenden Integrationstests verwenden die konfigurierte Datenbank.

Erstellen Sie 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"

Führe die Tests von der Projektwurzel aus:

pytest

Diese Tests verwenden die konfigurierte Datenbank und test_create_product fügen eine Zeile in SalesLT.Productein. Verwenden Sie eine dedizierte Testdatenbank und setzen Sie deren Daten zwischen den Testläufen zurück.

Checkliste für die Bereitstellung

  • Legen Sie DATABASE_SERVER, DATABASE_NAME und JWT_SECRET über die Secret- und Konfigurationsspeicher der Bereitstellungsplattform fest.
  • Verwenden Sie eine dedizierte Microsoft Entra-Identität mit den minimal erforderlichen Datenbankberechtigungen.
  • Stellen Sie die Poolgröße unter das Verbindungslimit der Datenbank ein und lassen Sie Kapazität für administrativen Zugriff und andere Arbeitslasten frei.
  • Führen Sie Datenbankintegrationstests gegen eine isolierte Testdatenbank durch.
  • Konfigurieren Sie geschützte Telemetrie für Datenbankausnahmen, Anfragelatenz und Pool-Erschöpfung.
  • Führe Uvicorn ohne --reload in deployierten Umgebungen aus.

Verwandte Inhalte

  • Last updated on