Tester et déployer des applications FastAPI avec mssql-python

Après avoir construit une application FastAPI avec mssql-python, configurez-la pour le déploiement, la réutilisation de connexions, la gestion des erreurs, l’authentification et les tests automatisés.

Prerequisites

• Utilisez MSSQL-python avec FastAPI, ou ayez une application FastAPI équivalente qui utilise la base de données d’exemple AdventureWorksLT. La dépendance d’authentification dans cet article interroge SalesLT.Customer.

• Installez les dépendances de production et de test :

  pip install pydantic-settings pyjwt pytest httpx
  

Configurer les paramètres de déploiement

Utilisez Paramètres Pydantic pour charger des valeurs spécifiques au déploiement à partir des variables d’environnement. Cette approche maintient les secrets hors du code source et offre à chaque environnement sa propre base de données, son pool et sa configuration d’authentification.

Créez 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"
    )

Définissez DATABASE_SERVER, DATABASE_NAME, et JWT_SECRET dans l’environnement de déploiement. Pydantic Settings lit automatiquement les noms de variables d’environnement en majuscules.

Note

ActiveDirectoryDefault essaie successivement plusieurs fournisseurs d’informations d’identification. En production, spécifiez le mode d’authentification de l’identité déployée, par exemple ActiveDirectoryMSI pour l’identité gérée, afin d’éviter de parcourir la chaîne des identifiants. Pour les modes disponibles, voir authentification Microsoft Entra avec mssql-python.

Configurer le regroupement de connexions

MSSQL-Python permet par défaut le pooling de connexions. Configurez le pool une fois, avant que l’application ne crée sa première connexion. Dimensionnez le pool pour le travail concurrent attendu de l’application dans la base de données et le niveau de service de la base de données.

Mise à jour database.py pour utiliser les paramètres de déploiement :

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

Le gestionnaire de contexte de connexion fait un commit après un traitement réussi des requêtes, revient en arrière lorsque le traitement des requêtes ouvre une exception, puis ferme la connexion. Fermer la connexion le ramène dans le pool. Pour les clés du pool, le dimensionnement, l’isolation des identités et les recommandations concernant l’épuisement, voir Pool de connexions avec mssql-python.

Gérer les erreurs de base de données

Enregistrez des gestionnaires d’exception afin que les défaillances de la base de données renvoient des réponses cohérentes sans exposer les détails de connexion, les requêtes ou le texte d’erreur du serveur.

Ajouter les manipulateurs après app = FastAPI(...) dans 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",
        },
    )

Enregistrez l’exception via le pipeline de télémétrie protégé de votre application avant de retourner la réponse. Pour la hiérarchie des exceptions et la gestion SQLSTATE, voir Gestion des erreurs et codes SQLSTATE pour mssql-python.

Ajouter des dépendances d’authentification

Enchaînez les dépendances FastAPI pour valider un JSON Web Token (JWT), charger le client AdventureWorksLT correspondant et mettre ce client à la disposition des routes protégées. Validez le jeton avant d’acquérir une connexion de base de données afin qu’un jeton invalide n’utilise pas une connexion poolée.

Créez 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,
    }

Importez la dépendance et ajoutez une route protégée vers main.py:

from auth import get_current_customer


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

Utilisez un fournisseur d’identité pour émettre et faire tourner les clés de signature. Pour HS256, réglez JWT_SECRET à au moins 32 octets aléatoires. Ne stockez pas un secret de signature de production dans le dépôt ou dans une image.

Tester l’application

FastAPI TestClient envoie des requêtes à l’application sans démarrer de serveur HTTP. Les tests d’intégration suivants utilisent la base de données configurée.

Créez 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"

Exécutez les tests à partir de la racine du projet :

pytest

Ces tests utilisent la base de données configurée, et test_create_product insèrent une ligne dans SalesLT.Product. Utilisez une base de données dédiée aux tests et réinitialisez ses données entre les essais.

Liste de vérification du déploiement

• Définir DATABASE_SERVER, DATABASE_NAME, et JWT_SECRET via les archives secrètes et de configuration de la plateforme de déploiement.
• Utilisez une identité Microsoft Entra dédiée avec les autorisations minimales requises pour la base de données.
• Réglez la taille du pool en dessous de la limite de connexion de la base de données et laissez une capacité pour l’accès administratif et d’autres charges de travail.
• Effectuez des tests d’intégration de base de données sur une base de données isolée.
• Configurez la télémétrie protégée pour les exceptions de base de données, la latence des requêtes et l’épuisement du pool.
• Utilisez Uvicorn sans --reload en environnement déployé.

Contenu connexe

• Utilisez mssql-python avec FastAPI
• Mise en commun de connexions avec mssql-python
• Gestion des erreurs et codes SQLSTATE pour mssql-python
• Authentification Microsoft Entra avec mssql-python
• Meilleures pratiques de sécurité pour les applications mssql-python