Usa mssql-python con FastAPI

FastAPI è un moderno framework web Python per la creazione di API. Combinato con mssql-python, puoi costruire API REST ad alte prestazioni supportate da Microsoft SQL e database SQL di Azure.

Prerequisiti

• Python 3.10 o versione successiva.
• I pacchetti mssql-python, fastapi, uvicorn, pydantic e PyJWT. Installa tutto con pip install fastapi uvicorn mssql-python pydantic pyjwt.
• Installare prerequisiti specifici del sistema operativo monouso. Gli utenti Windows possono saltare questo passaggio. Per dettagli completi sulla piattaforma, vedi Installa mssql-python.
  • Alpino
  • Debian/Ubuntu
  • RHEL
  • SUSE
  • openSUSE
  • macOS

  apk add libtool krb5-libs krb5-dev
  

Creare un database SQL

Crea o collegati a un database SQL su una delle seguenti piattaforme:

Gli esempi in questo articolo utilizzano il database di esempio AdventureWorksLT , in particolare la SalesLT.Product tabella. Se non hai installato AdventureWorksLT, consulta i database di esempio di AdventureWorks.

Configurazione del progetto

Creare un ambiente virtuale

Crea e attiva un ambiente virtuale in modo che i pacchetti di questo progetto rimangano isolati dalle altre installazioni Python. Questo passaggio previene anche il problema comune di installare pacchetti in un interprete mentre si esegue l'app o i test con un altro.

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

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

Dopo aver attivato l'ambiente, python, pip, e pytest tutti risolvono sullo stesso interprete. Esegui i comandi rimanenti in questo articolo dall'ambiente attivato.

Installa le dipendenze

Installa i pacchetti necessari con pip:

pip install fastapi uvicorn mssql-python pydantic pyjwt

Struttura del progetto

Organizza il tuo progetto con moduli separati per database, schemi e operazioni CRUD:

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

Gestione delle connessioni al database

FastAPI utilizza l'iniezione di dipendenza per fornire risorse come connessioni di database ai gestori di route. Il pattern in questa sezione crea un gestore di contesto che apre una connessione, genera un cursore e gestisce automaticamente commit/rollback/chiusura.

Crea database.py

La get_connection_string() funzione costruisce la stringa di connessione ODBC a partire dai valori di configurazione. Il get_db() gestore di contesto e il get_db_dependency() generatore seguono entrambi lo stesso schema: aprono una connessione, restituiscono un cursore, eseguono il commit in caso di esito positivo, eseguono il rollback in caso di errore e chiudono sempre la connessione al termine. FastAPI chiama Depends()get_db_dependency() una volta per richiesta e ne gestisce il ciclo di vita.

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

@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()

Modelli pidantici

I modelli pydantic definiscono le regole di forma e validazione per i dati di richiesta e risposta. FastAPI utilizza questi modelli per analizzare il JSON in ingresso, validare i vincoli del campo e generare automaticamente la documentazione OpenAPI.

Crea schemas.py

Separare gli schemi in Base, Create, Update, e varianti di risposta. Lo Base schema contiene i campi condivisi, Create ne eredita per le operazioni di inserimento e Update rende tutti i campi opzionali per gli aggiornamenti parziali.

# 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

Operazioni CRUD

Incapsula le query del database in una classe dedicata per mantenere i route handler sottili. Ogni metodo statico prende un cursore (iniettato da FastAPI) e gestisce un'operazione utilizzando query parametrizzate (%(name)s segnaposto con un dizionario di valori) per prevenire l'iniezione SQL. Questa separazione rende la logica di business più facile da testare e riutilizzare.

Crea 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()]

Applicazione FastAPI

Crea il file main.py

Il modulo principale collega tutto insieme. Ogni rotta dichiara cursor = Depends(get_db_dependency), che indica a FastAPI di chiamare il generatore, passare alla funzione handler il cursore restituito tramite yield ed eseguire poi la pulizia. FastAPI convalida anche i corpi delle richieste HTTP in base ai tuoi schemi Pydantic prima che l'handler venga eseguito.

# 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)}")

Eseguire l'applicazione

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

Gestione degli errori

FastAPI ti permette di registrare gestori globali di eccezioni per tipi specifici di eccezioni. Quando si catturano mssql_python.DatabaseError e mssql_python.IntegrityError, FastAPI restituisce errori JSON strutturati con codici di stato HTTP appropriati invece delle risposte generiche 500.

Gestore globale di eccezioni

Aggiungi questi handler a main.py, subito dopo la app = FastAPI(...) riga. FastAPI esegue l'handler corrispondente ogni volta che una route solleva quel tipo di eccezione, quindi non hai bisogno di un blocco try/except in ogni 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"}
    )

Pool di connessioni

Senza il pool di connessioni, ogni richiesta apre e chiude una connessione TCP a Microsoft SQL, che aggiunge latenza. Il pool di connessione mantiene un insieme di connessioni inattive pronte per il riutilizzo. Chiama mssql_python.pooling() una volta all'avvio. Con il pooling abilitato, conn.close()get_db_dependency() restituisce la connessione al pool invece di chiuderlo davvero.

Modulo database migliorato

Abilita il pooling chiamando mssql_python.pooling() all'avvio e configuralo con le impostazioni di dimensione massima e timeout appropriate:

# 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 per l'autenticazione

Puoi combinare l'accesso al database con l'autenticazione concatenando le dipendenze di FastAPI. Il seguente esempio valida un token portatore JWT, cerca il record persona corrispondente nel database di esempio AdventureWorksLT e rende il risultato disponibile per le rotte protette.

# 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 fornisce un TestClient strumento integrato basato su httpx per inviare richieste alla tua applicazione senza avviare un vero server HTTP. Scrivi test con pytest per verificare route, codici di stato e forme di risposta.

Prima di eseguire i test in questa sezione, installa le dipendenze di test:

pip install pytest httpx

Configurazione dei test

Crea un file di test che serve TestClient a verificare il comportamento delle rotte e gli schemi di risposta:

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

Esegui i test con pytest dalla radice del progetto, la stessa directory di main.py:

pytest

Questi test si eseguono sul tuo database live invece che sulle mock, quindi test_create_product inserisce una riga reale in SalesLT.Product. In AdventureWorksLT, sia Name che ProductNumber hanno vincoli unici, quindi il test genera un valore unico per ciascuno a ogni esecuzione. Se invece codifica quei valori in modo rigido, il test fallisce con un conflitto alla seconda esecuzione, a meno che tu non elimini prima la riga.

Configurazione della distribuzione

Usa il BaseSettings di Pydantic per caricare la configurazione dalle variabili d'ambiente e dai file .env. Questo approccio tiene i segreti fuori dal codice sorgente e rende facile passare da un ambiente all'altro. Installa il pacchetto impostazioni con pip install pydantic-settings.

Variabili di ambiente

Crea un modulo di impostazioni che carichi la configurazione dalle variabili dell'ambiente, permettendoti di gestire segreti e valori specifici per la distribuzione al di fuori del tuo codice:

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

Poi aggiorna database.py per importare get_connection_string da config invece di definire una propria copia. Rimuovendo la funzione duplicata, ti assicuri che l'app legga le impostazioni di connessione da una singola fonte.

# database.py
from config import get_connection_string

Contenuti correlati

• Gestione delle connessioni
• Pool di connessioni
• Gestione degli errori
• Autenticazione Microsoft Entra