Utilisez mssql-python avec FastAPI

FastAPI est un framework web Python moderne pour la création d’API. Combiné à mssql-python, vous pouvez construire des API REST haute performance soutenues par Microsoft SQL et Azure SQL Database.

Logiciels requis

  • Python 3.10 ou version ultérieure.
  • Les packages mssql-python, fastapi, uvicorn, pydantic et PyJWT. Installez tout avec pip install fastapi uvicorn mssql-python pydantic pyjwt.
  • Installez les prérequis ponctuels spécifiques au système d'exploitation. Les utilisateurs Windows peuvent sauter cette étape. Pour tous les détails sur la plateforme, voir Installer mssql-python.
    apk add libtool krb5-libs krb5-dev

Créer une base de données SQL

Créer ou connecter une base de données SQL sur l’une des plateformes suivantes :

Les exemples de cet article utilisent la base de données d’exemples AdventureWorksLT , en particulier la SalesLT.Product table. Si vous n’avez pas AdventureWorksLT installé, consultez les bases de données d’exemple AdventureWorks.

Configuration du projet

Créer un environnement virtuel

Créez et activez un environnement virtuel afin que les paquets de ce projet restent isolés des autres installations Python. Cette étape évite également le problème courant d’installer des paquets dans un interpréteur lors de l’exécution de votre application ou de tests avec un autre.

py -m venv .venv
.\.venv\Scripts\Activate.ps1

Après avoir activé l’environnement, python, pip et pytest pointent tous vers le même interpréteur. Exécutez les commandes restantes de cet article depuis l’environnement activé.

Note

Sous Windows sur Arm, créez l’environnement avec un build Arm64 de Python afin que mssql-python et ses dépendances s’installent à partir de wheels précompilés. Sur une machine où plusieurs versions de Python sont installées, py -m venv peut sélectionner une version ou une architecture différente de celle attendue ; vérifiez avec python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())" après avoir activé. Si pip tente de compiler cryptography depuis les sources (erreur liée à la chaîne d’outils Rust et OpenSSL), installez d’abord une version fournie sous forme de wheel avec pip install --only-binary=:all: cryptography, puis installez le reste.

Installer des dépendances

Installez les packages requis avec pip :

pip install fastapi uvicorn mssql-python pydantic pyjwt

Structure du projet

Organisez votre projet avec des modules distincts pour la base de données, les schémas et les opérations CRUD :

my_api/
├── main.py
├── database.py
├── models.py
├── schemas.py
├── crud.py
├── test_api.py
└── routers/
    └── products.py

Gestion des connexions à la base de données

FastAPI utilise l’injection de dépendances pour fournir des ressources comme les connexions de bases de données aux gestionnaires de routes. Le modèle présenté dans cette section crée un gestionnaire de contexte qui ouvre une connexion, fournit un curseur et gère automatiquement la validation, l’annulation ou la fermeture.

Créez database.py

La get_connection_string() fonction construit la chaîne de connexion ODBC à partir de valeurs de configuration. Le get_db() gestionnaire de contexte et le get_db_dependency() générateur suivent tous deux le même schéma : ouvrir une connexion, produire un curseur, valider la transaction en cas de succès, annuler la transaction en cas d’erreur, et toujours fermer la connexion une fois l’opération terminée. FastAPI appelle Depends()get_db_dependency() une fois par requête et gère son cycle de vie.

# database.py
import mssql_python
from contextlib import contextmanager
from typing import Generator

# Configuration
DATABASE_CONFIG = {
    "server": "<server>.database.windows.net",
    "database": "<database>",
}

def get_connection_string() -> str:
    """Build connection string from config."""
    return (
        f"Server={DATABASE_CONFIG['server']};"
        f"Database={DATABASE_CONFIG['database']};"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes"
    )

Note

ActiveDirectoryDefault utilise DefaultAzureCredential, qui tente successivement d’utiliser plusieurs fournisseurs d’informations d’identification. La première connexion peut être lente car le SDK parcourt la chaîne jusqu’à ce qu’il trouve un fournisseur fonctionnel. En production, si vous savez quel type d’identifiant votre environnement utilise, spécifiez-le directement (par exemple, ActiveDirectoryMSI pour l’identité gérée) afin d’éviter la marche en chaîne. Pour plus d’informations, consultez Authentification Microsoft Entra.

@contextmanager
def get_db() -> Generator:
    """Database connection context manager for FastAPI dependency injection."""
    conn = mssql_python.connect(get_connection_string())
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()

def get_db_dependency():
    """FastAPI dependency for database cursor."""
    conn = mssql_python.connect(get_connection_string())
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()

Modèles pydantiques

Les modèles pydantiques définissent la forme et les règles de validation pour les données de requête et de réponse. FastAPI utilise ces modèles pour analyser le JSON entrant, valider les contraintes de champ et générer automatiquement la documentation OpenAPI.

Créez schemas.py

Séparer les schémas en Base, Create, Update, et répondre aux variantes. Le Base schéma contient des champs partagés, Create en hérite pour les opérations d’insertion, et Update rend tous les champs optionnels pour les mises à jour partielles.

# schemas.py
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from typing import Optional
from datetime import datetime

# Product schemas
class ProductBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    product_number: str = Field(..., min_length=1, max_length=25)
    price: float = Field(..., gt=0)
    color: Optional[str] = Field(None, max_length=50)
    size: Optional[str] = Field(None, max_length=50)
    category_id: Optional[int] = None

class ProductCreate(ProductBase):
    pass

class ProductUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    product_number: Optional[str] = Field(None, min_length=1, max_length=25)
    price: Optional[float] = Field(None, gt=0)
    color: Optional[str] = Field(None, max_length=50)
    size: Optional[str] = Field(None, max_length=50)
    category_id: Optional[int] = None

class Product(ProductBase):
    id: int

    model_config = ConfigDict(from_attributes=True)

# Pagination
class PaginatedResponse(BaseModel):
    items: list
    total: int
    page: int
    page_size: int
    pages: int

Opérations CRUD

Encapsulez les requêtes de base de données dans une classe dédiée afin d’alléger les gestionnaires de routes. Chaque méthode statique prend un curseur (injecté par FastAPI) et gère une opération à l’aide de requêtes paramétrées (%(name)s placeholders avec un dictionnaire de valeurs) afin d’empêcher l’injection SQL. Cette séparation facilite les tests et la réutilisation de la logique métier.

Créez crud.py

# crud.py
from typing import Optional, List
from schemas import ProductCreate, ProductUpdate, Product

class ProductCRUD:
    """CRUD operations for products."""
    
    @staticmethod
    def get(cursor, product_id: int) -> Optional[dict]:
        cursor.execute("""
            SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
            FROM SalesLT.Product
            WHERE ProductID = %(id)s
        """, {"id": product_id})
        
        row = cursor.fetchone()
        if row:
            return {
                "id": row.ProductID,
                "name": row.Name,
                "product_number": row.ProductNumber,
                "price": float(row.ListPrice),
                "color": row.Color,
                "size": row.Size
            }
        return None
    
    @staticmethod
    def get_all(cursor, skip: int = 0, limit: int = 100) -> List[dict]:
        cursor.execute("""
            SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
            FROM SalesLT.Product
            ORDER BY ProductID
            OFFSET %(skip)s ROWS
            FETCH NEXT %(limit)s ROWS ONLY
        """, {"skip": skip, "limit": limit})
        
        return [{
            "id": row.ProductID,
            "name": row.Name,
            "product_number": row.ProductNumber,
            "price": float(row.ListPrice),
            "color": row.Color,
            "size": row.Size
        } for row in cursor.fetchall()]
    
    @staticmethod
    def count(cursor) -> int:
        cursor.execute("SELECT COUNT(*) FROM SalesLT.Product")
        return cursor.fetchval()
    
    @staticmethod
    def create(cursor, product: ProductCreate) -> dict:
        cursor.execute("""
            INSERT INTO SalesLT.Product (Name, ProductNumber, ListPrice, Color, Size, ProductCategoryID, StandardCost, SellStartDate)
            OUTPUT INSERTED.ProductID, INSERTED.Name, INSERTED.ProductNumber,
                   INSERTED.ListPrice, INSERTED.Color, INSERTED.Size
            VALUES (%(name)s, %(product_number)s, %(price)s, %(color)s, %(size)s, %(category_id)s, 0, GETDATE())
        """, {
            "name": product.name,
            "product_number": product.product_number,
            "price": product.price,
            "color": product.color,
            "size": product.size,
            "category_id": product.category_id
        })
        
        row = cursor.fetchone()
        return {
            "id": row.ProductID,
            "name": row.Name,
            "product_number": row.ProductNumber,
            "price": float(row.ListPrice),
            "color": row.Color,
            "size": row.Size
        }
    
    @staticmethod
    def update(cursor, product_id: int, product: ProductUpdate) -> Optional[dict]:
        # Build dynamic update
        updates = []
        params = {"id": product_id}
        
        if product.name is not None:
            updates.append("Name = %(name)s")
            params["name"] = product.name
        if product.product_number is not None:
            updates.append("ProductNumber = %(product_number)s")
            params["product_number"] = product.product_number
        if product.price is not None:
            updates.append("ListPrice = %(price)s")
            params["price"] = product.price
        if product.category_id is not None:
            updates.append("ProductCategoryID = %(category_id)s")
            params["category_id"] = product.category_id
        
        if not updates:
            return ProductCRUD.get(cursor, product_id)
        
        cursor.execute(f"""
            UPDATE SalesLT.Product SET {', '.join(updates)}
            OUTPUT INSERTED.ProductID, INSERTED.Name, INSERTED.ProductNumber,
                   INSERTED.ListPrice, INSERTED.Color, INSERTED.Size
            WHERE ProductID = %(id)s
        """, params)
        
        row = cursor.fetchone()
        if row:
            return {
                "id": row.ProductID,
                "name": row.Name,
                "product_number": row.ProductNumber,
                "price": float(row.ListPrice),
                "color": row.Color,
                "size": row.Size
            }
        return None
    
    @staticmethod
    def delete(cursor, product_id: int) -> bool:
        cursor.execute("""
            DELETE FROM SalesLT.Product WHERE ProductID = %(id)s
        """, {"id": product_id})
        return cursor.rowcount > 0
    
    @staticmethod
    def search(cursor, query: str, skip: int = 0, limit: int = 100) -> List[dict]:
        cursor.execute("""
            SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
            FROM SalesLT.Product
            WHERE Name LIKE %(query)s OR ProductNumber LIKE %(query)s
            ORDER BY ProductID
            OFFSET %(skip)s ROWS
            FETCH NEXT %(limit)s ROWS ONLY
        """, {"query": f"%{query}%", "skip": skip, "limit": limit})
        
        return [{
            "id": row.ProductID,
            "name": row.Name,
            "product_number": row.ProductNumber,
            "price": float(row.ListPrice),
            "color": row.Color,
            "size": row.Size
        } for row in cursor.fetchall()]

Application FastAPI

Créez main.py

Le module principal relie tout ensemble. Chaque route déclare cursor = Depends(get_db_dependency), ce qui indique à FastAPI d’appeler le générateur, de transmettre au gestionnaire le curseur produit, puis d’effectuer le nettoyage. FastAPI valide également les corps de requêtes par rapport à vos schémas Pydantic avant que le gestionnaire ne s’exécute.

# main.py
from fastapi import FastAPI, HTTPException, Depends, Query
from typing import List
from database import get_db_dependency
from schemas import Product, ProductCreate, ProductUpdate, PaginatedResponse
from crud import ProductCRUD

app = FastAPI(
    title="Product API",
    description="REST API for products using mssql-python",
    version="1.0.0"
)

@app.get("/")
def root():
    return {"message": "Product API", "docs": "/docs"}

@app.get("/products", response_model=PaginatedResponse)
def list_products(
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    cursor = Depends(get_db_dependency)
):
    """List all products with pagination."""
    skip = (page - 1) * page_size
    items = ProductCRUD.get_all(cursor, skip=skip, limit=page_size)
    total = ProductCRUD.count(cursor)
    
    return {
        "items": items,
        "total": total,
        "page": page,
        "page_size": page_size,
        "pages": (total + page_size - 1) // page_size
    }

@app.get("/products/{product_id}", response_model=Product)
def get_product(product_id: int, cursor = Depends(get_db_dependency)):
    """Get a specific product by ID."""
    product = ProductCRUD.get(cursor, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return product

@app.post("/products", response_model=Product, status_code=201)
def create_product(product: ProductCreate, cursor = Depends(get_db_dependency)):
    """Create a new product."""
    return ProductCRUD.create(cursor, product)

@app.put("/products/{product_id}", response_model=Product)
def update_product(
    product_id: int,
    product: ProductUpdate,
    cursor = Depends(get_db_dependency)
):
    """Update an existing product."""
    updated = ProductCRUD.update(cursor, product_id, product)
    if not updated:
        raise HTTPException(status_code=404, detail="Product not found")
    return updated

@app.delete("/products/{product_id}", status_code=204)
def delete_product(product_id: int, cursor = Depends(get_db_dependency)):
    """Delete a product."""
    if not ProductCRUD.delete(cursor, product_id):
        raise HTTPException(status_code=404, detail="Product not found")

@app.get("/products/search/", response_model=List[Product])
def search_products(
    q: str = Query(..., min_length=1),
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    cursor = Depends(get_db_dependency)
):
    """Search products by name or product number."""
    skip = (page - 1) * page_size
    return ProductCRUD.search(cursor, q, skip=skip, limit=page_size)

# Health check endpoint
@app.get("/health")
def health_check(cursor = Depends(get_db_dependency)):
    """Check database connectivity."""
    try:
        cursor.execute("SELECT 1")
        return {"status": "healthy", "database": "connected"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"Database unhealthy: {str(e)}")

Exécuter l’application

uvicorn main:app --reload --host 0.0.0.0 --port 8000

Gestion des erreurs

FastAPI permet d’enregistrer des gestionnaires d’exceptions globaux pour des types d’exceptions spécifiques. Lorsque vous détectez mssql_python.DatabaseError et mssql_python.IntegrityError, FastAPI renvoie des erreurs JSON structurées avec des codes d’état HTTP appropriés au lieu de réponses génériques 500.

Gestionnaire d’exception global

Ajoutez ces gestionnaires à main.py, juste après la ligne app = FastAPI(...). FastAPI exécute le gestionnaire correspondant chaque fois qu’une route lève une exception de ce type, vous n’avez donc pas besoin d’un bloc try/except dans chaque route.

# main.py
from fastapi import Request
from fastapi.responses import JSONResponse
import mssql_python

@app.exception_handler(mssql_python.DatabaseError)
async def database_exception_handler(request: Request, exc: mssql_python.DatabaseError):
    """Handle database errors globally."""
    return JSONResponse(
        status_code=500,
        content={"detail": "Database error occurred", "type": "database_error"}
    )

@app.exception_handler(mssql_python.IntegrityError)
async def integrity_exception_handler(request: Request, exc: mssql_python.IntegrityError):
    """Handle integrity constraint violations."""
    error_msg = str(exc)
    
    if "UNIQUE" in error_msg:
        return JSONResponse(
            status_code=409,
            content={"detail": "Resource already exists", "type": "duplicate_error"}
        )
    elif "FOREIGN KEY" in error_msg:
        return JSONResponse(
            status_code=400,
            content={"detail": "Referenced resource not found", "type": "reference_error"}
        )
    
    return JSONResponse(
        status_code=400,
        content={"detail": "Data integrity error", "type": "integrity_error"}
    )

Note

La suppression d’un produit auquel d’autres lignes font encore référence déclenche mssql_python.IntegrityError en raison de la contrainte de clé étrangère, et le gestionnaire renvoie un code 400 au lieu de supprimer la ligne. Dans l’exemple AdventureWorksLT, la plupart des produits de SalesLT.Product sont référencés par SalesLT.SalesOrderDetail, donc DELETE échoue dans leur cas par conception. Pour tester une suppression réussie, créez un produit avec POST /products et supprimez celui-ci, ou supprimez d’abord les lignes de référencement.

Regroupement de connexions

Sans pooling de connexions, chaque requête ouvre et ferme une connexion TCP vers Microsoft SQL, ce qui ajoute de la latence. Le pooling de connexions maintient un ensemble de connexions inactives prêtes à être réutilisées. Appelez mssql_python.pooling() une fois au démarrage. Lorsque la mise en pool est activée, conn.close() dans get_db_dependency() renvoie la connexion au pool de connexions au lieu de fermer réellement la connexion.

Module de base de données amélioré

Activez le pooling en appelant mssql_python.pooling() au démarrage et configurez-le avec des réglages de taille maximale et de délai appropriés :

# database.py with connection pooling
import mssql_python
from contextlib import contextmanager
import os

# Configure pool
mssql_python.pooling(max_size=20, idle_timeout=300)

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "Server=<server>.database.windows.net;Database=<database>;"
    "Authentication=ActiveDirectoryDefault;Encrypt=yes"
)

def get_db_dependency():
    """FastAPI dependency with connection pooling."""
    conn = mssql_python.connect(DATABASE_URL)
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()  # Returns to pool

Middleware d’authentification

Vous pouvez combiner l’accès à la base de données avec l’authentification en chaînant les dépendances FastAPI. L’exemple suivant valide un jeton JWT Bearer, recherche la fiche de la personne correspondante dans la base de données d’exemple AdventureWorksLT, puis met le résultat à la disposition des routes protégées.

# auth.py
from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer()

def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    cursor = Depends(get_db_dependency)
):
    """Validate JWT and return the matching AdventureWorksLT person."""
    try:
        token = credentials.credentials
        # Replace with a strong secret loaded from environment variables
        payload = jwt.decode(token, "your-secret-key", algorithms=["HS256"])
        person_id = int(payload.get("sub"))
        
        if not person_id:
            raise HTTPException(status_code=401, detail="Invalid token")
        
        cursor.execute("""
            SELECT BusinessEntityID, FirstName, LastName
            FROM Person.Person
            WHERE BusinessEntityID = %(id)s
        """, {"id": person_id})
        
        person = cursor.fetchone()
        if not person:
            raise HTTPException(status_code=401, detail="User not found")
        
        return {
            "id": person.BusinessEntityID,
            "first_name": person.FirstName,
            "last_name": person.LastName
        }
        
    except (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")

# Protected endpoint
@app.get("/me")
def get_me(current_user: dict = Depends(get_current_user)):
    return current_user

Testing

FastAPI propose un TestClient module intégré httpx qui envoie des requêtes à votre application sans démarrer un vrai serveur HTTP. Écrivez des tests avec pytest pour vérifier les routes, les codes d’état et la structure des réponses.

Avant d’exécuter les tests de cette section, installez les dépendances de test :

pip install pytest httpx

Note

Si vous utilisez la dernière version de Starlette ou si vous mettez en place un nouvel environnement, préférez httpx2 à httpx. Les versions récentes de Starlette utilisent httpx2 pour TestClient et émettent un avertissement de dépréciation lorsque seule httpx est installée. Installez-le avec pip install pytest httpx2.

Configuration de test

Créez un fichier de test qui sert TestClient à vérifier le comportement des routes et les schémas de réponse :

# test_api.py
from fastapi.testclient import TestClient
from main import app
import uuid
import pytest

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]
    name = f"Test Product {suffix}"
    product_data = {
        "name": name,
        "product_number": f"TEST-{suffix}",
        "price": 19.99,
        "color": "Red",
        "size": "M",
        "category_id": 1
    }
    response = client.post("/products", json=product_data)
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == name
    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 avec pytest depuis la racine du projet, le même répertoire que main.py:

pytest

Ces tests s’exécutent sur votre base de données en direct plutôt que sur des mocks, donc test_create_product insère une vraie ligne dans SalesLT.Product. Dans AdventureWorksLT, Name et ProductNumber ont des contraintes uniques, donc le test génère une valeur unique pour chacun à chaque exécution. Si vous codez ces valeurs en dur à la place, le test échoue avec un conflit lors de la deuxième exécution, sauf si vous supprimez d’abord la ligne.

Configuration du déploiement

Utilisez Pydantic BaseSettings pour charger la configuration à partir des variables et .env fichiers d’environnement. Cette approche empêche les secrets d’entrer dans le code source et facilite le passage d’un environnement à l’autre. Installez le package de paramètres avec pip install pydantic-settings.

Variables d’environnement

Créez un module de paramètres qui charge la configuration à partir des variables d’environnement, vous permettant de gérer des secrets et des valeurs spécifiques au déploiement en dehors de votre code :

# config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    database_server: str = "<server>.database.windows.net"
    database_name: str = "<database>"
    pool_size: int = 10

    model_config = SettingsConfigDict(env_file=".env")

settings = Settings()

def get_connection_string() -> str:
    return (
        f"Server={settings.database_server};"
        f"Database={settings.database_name};"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes"
    )

Puis mettre à jour database.py pour importer get_connection_string depuis config au lieu de définir sa propre copie. En supprimant la fonction dupliquée, vous vous assurez que l’application lit les paramètres de connexion d’une seule source.

# database.py
from config import get_connection_string

Contenu connexe

  • Last updated on