Testare e distribuire applicazioni FastAPI con mssql-python

Dopo aver creato un'applicazione FastAPI con mssql-python, configurala per il deployment, il riutilizzo delle connessioni, la gestione degli errori, l'autenticazione e i test automatizzati.

Prerequisites

• Completa Usa mssql-python con FastAPI, oppure hai un'applicazione FastAPI equivalente che utilizza il database di esempio AdventureWorksLT. La dipendenza di autenticazione in questo articolo interroga SalesLT.Customer.

• Installa le dipendenze di produzione e test:

  pip install pydantic-settings pyjwt pytest httpx
  

Configurare le impostazioni di distribuzione

Usa Pydantic Settings per caricare valori specifici di distribuzione dalle variabili dell'ambiente. Questo approccio tiene i segreti fuori dal codice sorgente e fornisce a ciascun ambiente il proprio database, pool e configurazione di autenticazione.

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

Imposta DATABASE_SERVER, DATABASE_NAME, e JWT_SECRET nell'ambiente di distribuzione. Pydantic Settings legge automaticamente i nomi delle variabili ambientali maiuscole.

Note

ActiveDirectoryDefault Prova più fornitori di credenziali in sequenza. In produzione, specificare la modalità di autenticazione per l'identità distribuita, ad esempio ActiveDirectoryMSI per l'identità gestita, per evitare di percorrere la catena delle credenziali. Per le modalità disponibili, vedi Microsoft Entra authentication with mssql-python.

Configurare il pool di connessioni

MSSQL-Python abilita di default il pool di connessioni. Configura il pool una volta, prima che l'applicazione crei la sua prima connessione. Dimensiona il pool in base al carico di lavoro simultaneo previsto sul database dell'applicazione e al livello di servizio del database.

Aggiorna database.py per utilizzare le impostazioni di distribuzione:

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

Il gestore del contesto della connessione effettua il commit dopo l'elaborazione delle richieste con successo, torna indietro quando l'elaborazione delle richieste solleva un'eccezione e chiude la connessione. Chiudendo la connessione la riporta al pool. Per informazioni su chiavi del pool, dimensionamento, isolamento delle identità e indicazioni sull'esaurimento, vedi Pooling delle connessioni con mssql-python.

Gestione degli errori del database

Registrare i gestori di eccezioni affinché i guasti del database restituiscano risposte coerenti senza esporre dettagli di connessione, query o testo di errore del server.

Aggiungi gli handler dopo app = FastAPI(...) in 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",
        },
    )

Registra l'eccezione attraverso la pipeline di telemetria protetta della tua applicazione prima di restituire la risposta. Per la gerarchia delle eccezioni e la gestione SQLSTATE, vedi Gestione errori e codici SQLSTATE per mssql-python.

Aggiungi dipendenze di autenticazione

Concatena le dipendenze di FastAPI per convalidare un JSON Web Token (JWT), caricare il cliente AdventureWorksLT corrispondente e rendere il cliente disponibile alle route protette. Valida il token prima di acquisire una connessione al database così un token non valido non utilizza una connessione pooled.

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

Importare la dipendenza e aggiungere una via protetta verso main.py:

from auth import get_current_customer


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

Usa un fornitore di identità per emettere e ruotare le chiavi di firma. Per usare HS256, imposta JWT_SECRET ad almeno 32 byte casuali. Non memorizzare un segreto di firma di produzione nel repository o in un'immagine.

Testare l'applicazione

FastAPI TestClient invia richieste all'applicazione senza avviare un server HTTP. I seguenti test di integrazione utilizzano il database configurato.

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

Esegui i test dalla radice del progetto:

pytest

Questi test utilizzano il database configurato e test_create_product inseriscono una riga in SalesLT.Product. Usa un database di test dedicato e resetta i dati tra una run e l'altra.

Elenco di controllo per la distribuzione

• Imposta DATABASE_SERVER, DATABASE_NAME, e JWT_SECRET tramite i record segreti e di configurazione della piattaforma di distribuzione.
• Usa un'identità Microsoft Entra dedicata con i permessi minimi richiesti.
• Imposta la dimensione del pool al di sotto del limite di connessione del database e lascia la capacità per l'accesso amministrativo e altri carichi di lavoro.
• Esegui test di integrazione del database su un database di test isolato.
• Configura la telemetria protetta per le eccezioni del database, la latenza delle richieste e l'esaurimento del pool.
• Usa Uvicorn senza --reload negli ambienti di distribuzione.

Contenuti correlati

• Usa mssql-python con FastAPI
• Pooling delle connessioni con mssql-python
• Gestione degli errori e codici SQLSTATE per mssql-python
• Autenticazione Microsoft Entra con mssql-python
• Migliori pratiche di sicurezza per applicazioni mssql-python