Testa och distribuera FastAPI-applikationer med mssql-python

Efter att du byggt en FastAPI-applikation med mssql-python, konfigurera den för distribution, återanvändning av anslutningar, felhantering, autentisering och automatiserad testning.

Förutsättningar

• Fullständigt Använd mssql-python med FastAPI, eller ha en motsvarande FastAPI-applikation som använder AdventureWorksLT:s exempeldatabas. Beroendet för autentisering i den här artikeln frågar efter SalesLT.Customer.

• Installera produktions- och testberoenden:

  pip install pydantic-settings pyjwt pytest httpx
  

Konfigurera inställningar för distribution

Använd Pydantic Settings för att ladda deployment-specifika värden från miljövariabler. Denna metod håller hemligheter utanför källkoden och ger varje miljö sin egen databas, pool och autentiseringskonfiguration.

Skapa 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"
    )

Sätt DATABASE_SERVER, DATABASE_NAME, och JWT_SECRET i distributionsmiljön. Pydantic Settings läser automatiskt in miljövariabelnamn skrivna med versaler.

Konfigurera anslutningspooler

MSSQL-Python möjliggör anslutningspoolning som standard. Konfigurera poolen en gång, innan applikationen skapar sin första anslutning. Storlek på poolen för applikationens förväntade samtidiga databasarbete och databasservicenivån.

Uppdatera database.py för att använda distributionsinställningarna:

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

Kontexthanteraren för anslutningen bekräftar efter att begäran har behandlats utan fel, rullar tillbaka om behandlingen av begäran utlöser ett undantag, och stänger anslutningen. Att stänga anslutningen returnerar den till poolen. För poolnycklar, dimensionering, identitetsisolering och vägledning om uttömning, se Anslutningspoolning med mssql-python.

Hantera databasfel

Registrera undantagshanterare så att databasfel returnerar konsekventa svar utan att exponera anslutningsdetaljer, frågor eller serverfeltext.

Lägg till hanterarna efter app = FastAPI(...) i 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",
        },
    )

Logga undantaget genom applikationens skyddade telemetripipeline innan du returnerar svaret. För undantagshierarkin och SQLSTATE-hantering, se Felhantering och SQLSTATE-koder för mssql-python.

Lägg till autentiseringsberoenden

Kedja ihop FastAPI-beroenden för att validera en JSON Web Token (JWT), läsa in motsvarande AdventureWorksLT-kund och göra kunden tillgänglig för skyddade rutter. Validera token innan du skaffar en databasanslutning så att en ogiltig token inte använder en poolad anslutning.

Skapa 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,
    }

Importera beroendet och lägg till en skyddad rutt till main.py:

from auth import get_current_customer


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

Använd en identitetsleverantör för att utfärda och rotera signeringsnycklar. För HS256, sätt JWT_SECRET till minst 32 slumpmässiga byte. Spara inte en produktionssigneringshemlighet i arkivet eller i en bild.

Testa programmet

FastAPI TestClient skickar förfrågningar till applikationen utan att starta en HTTP-server. Följande integrationstester använder den konfigurerade databasen.

Skapa 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"

Kör testerna från projektets rot:

pytest

Dessa tester använder den konfigurerade databasen och test_create_product infogar en rad i SalesLT.Product. Använd en dedikerad testdatabas och återställ dess data mellan testkörningarna.

Checklista för distribution

• Ange DATABASE_SERVER, DATABASE_NAME och JWT_SECRET via driftsättningsplattformens hemlighetshantering och konfigurationslagring.
• Använd en dedikerad Microsoft Entra-identitet med de minsta nödvändiga databasbehörigheterna.
• Ställ in poolstorleken under databasens anslutningsgräns och lämna kapacitet för administrativ åtkomst och andra arbetsbelastningar.
• Kör databasintegrationstester mot en isolerad testdatabas.
• Konfigurera skyddad telemetri för databasundantag, förfrågningslatens och poolutmattning.
• Kör Uvicorn utan --reload i deployerade miljöer.

Relaterat innehåll

• Använd mssql-python med FastAPI
• Anslutningspoolning med mssql-python
• Felhantering och SQLSTATE-koder för mssql-python
• Microsoft Entra-autentisering med mssql-python
• Säkerhetsbästa praxis för mssql-python-applikationer