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.

Prerequisites

• Python 3.10 ou version ultérieure.
• 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.
  Alpin

  apk add libtool krb5-libs krb5-dev
  

  Debian/Ubuntu

  apt-get install -y libltdl7 libkrb5-3 libgssapi-krb5-2
  

  RHEL

  dnf install -y libtool-ltdl krb5-libs
  

  SUSE

  zypper install -y libltdl7 libkrb5-3 libgssapi-krb5-2
  

  openSUSE

  zypper install -y libltdl7
  

  macOS

  brew install openssl
  

Créer une base de données SQL

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

Base de données SQL Azure

Créer une base de données SQL en quelques minutes à l’aide du portail Azure

Base de données SQL dans Fabric

Charger des exemples de données AdventureWorks dans votre base de données SQL dans Microsoft Fabric

Microsoft SQL Server

Installer SQL Server depuis l’Assistant d’installation

Exemples de bases de données AdventureWorks

Conteneur SQL Server

Créez un conteneur SQL Server dans Docker

Créer un conteneur SQL Server avec sqlcmd

Créer un conteneur SQL Server avec l’extension MSSQL pour VS Code

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.

Windows

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

Linux/macOS

python3 -m venv .venv
source .venv/bin/activate

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 la source (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

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
└── 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 motif de cette section ouvre une connexion, génère un curseur, et utilise le gestionnaire de contexte de connexion mssql-python pour valider en cas de réussite, revenir en arrière sur une exception, puis fermer la connexion.

Créez database.py

La get_connection_string() fonction construit la chaîne de connexion ODBC à partir de valeurs de configuration. FastAPI appelle Depends()get_db_dependency() une fois par requête et gère son cycle de vie.

# database.py
import mssql_python
from collections.abc 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.

def get_db_dependency() -> Generator:
    """FastAPI dependency for database cursor."""
    with mssql_python.connect(get_connection_string()) as conn:
        with conn.cursor() as cursor:
            yield cursor

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:
        raise HTTPException(status_code=503, detail="Database unavailable")

Exécuter l’application

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

Tester et déployer l’application

Utilisez l’article complémentaire pour finaliser la demande :

Gestion des erreurs

L’article complémentaire traite de la gestion des exceptions dans les bases de données.

Gestionnaire d’exception global

Voir Gérer les erreurs de base de données.

Regroupement de connexions

L’article compagnon traite de la configuration du pool de connexions.

Module de base de données amélioré

Voir Configurer le pooling de connexions.

Middleware d’authentification

Voir Ajouter des dépendances d’authentification.

Testing

L’article complémentaire couvre les tests d’intégration.

Configuration de test

Voir Tester l’application.

Configuration du déploiement

L’article complémentaire traite de la configuration et des opérations de déploiement.

Variables d’environnement

Voir Configurer les paramètres de déploiement et la liste de contrôle du déploiement.

Contenu connexe

• Tester et déployer des applications FastAPI avec mssql-python
• Gérer les connexions avec mssql-python
• Mise en commun de connexions avec mssql-python
• Gestion des erreurs et codes SQLSTATE pour mssql-python
• Authentification Microsoft Entra avec mssql-python