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.

Installare prerequisiti specifici del sistema operativo monouso. Gli utenti Windows possono saltare questo passaggio. Per dettagli completi sulla piattaforma, vedi Installa mssql-python.

apk add libtool krb5-libs krb5-dev

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

dnf install -y libtool-ltdl krb5-libs

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

zypper install -y libltdl7

brew install openssl

Creare un database SQL

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

Creare un database SQL in pochi minuti usando il portale di Azure

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.

Windows
Linux/macOS

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.

Note

In Windows on Arm, crea l'ambiente con una build Arm64 di Python in modo che mssql-python e le relative dipendenze vengano installati da wheel precompilate. Su un computer con più versioni di Python, py -m venv potrebbe selezionare una versione o un'architettura diversa da quella prevista, quindi verifica con python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())" dopo l'attivazione. Se pip prova a compilare cryptography dal codice sorgente (un errore di toolchain di Rust e OpenSSL), installa prima una versione con ruota con pip install --only-binary=:all: cryptography, poi installa il resto.

Installa le dipendenze

Installa i pacchetti necessari con pip:

pip install fastapi uvicorn mssql-python pydantic

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
└── 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 modello in questa sezione apre una connessione, restituisce un cursore e utilizza il context manager della connessione di mssql-python per eseguire il commit in caso di esito positivo, eseguire il rollback in caso di eccezione e chiudere la connessione.

Crea database.py

La get_connection_string() funzione costruisce la stringa di connessione ODBC a partire dai valori di configurazione. FastAPI chiama Depends()get_db_dependency() una volta per richiesta e ne gestisce il ciclo di vita.

# 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 usa DefaultAzureCredential, che prova più fornitori di credenziali in sequenza. La prima connessione può essere lenta perché l'SDK percorre la catena finché non trova un fornitore funzionante. In produzione, se sai quale tipo di credenziale utilizza il tuo ambiente, specificalo direttamente (ad esempio, ActiveDirectoryMSI per l'identità gestita) per evitare il chain walk. Per altre informazioni, vedere Autenticazione di 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

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

Eseguire l'applicazione

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

Testare e distribuire l'applicazione

Usa l'articolo complementare per completare la domanda:

Gestione degli errori

L'articolo complementare tratta la gestione delle eccezioni nel database.

Gestore globale di eccezioni

Vedi Gestire gli errori del database.

Pool di connessioni

L'articolo complementare tratta la configurazione del pool di connessione.

Modulo database migliorato

Vedi Configura il pool di connessioni.

Middleware per l'autenticazione

Vedi Aggiungi dipendenze di autenticazione.

Testing

L'articolo complementare tratta i test di integrazione.

Configurazione dei test

Vedi Testare l'applicazione.

Configurazione della distribuzione

L'articolo complementare tratta la configurazione e le operazioni del dispiegamento.

Variabili di ambiente

Vedi Configura impostazioni di distribuzione e la checklist per il deployment.

Commenti e suggerimenti

Questa pagina è stata utile?

Last updated on 2026-09-23