FastAPI non e un semplice wrapper attorno a Starlette. La combinazione di Pydantic v2 per la validazione, Depends() per la dependency injection e la gestione delle richieste async-native lo rende il piu vicino equivalente Python a un framework HTTP tipizzato. La versione originale di questo articolo aveva una firma dell’handler errata e nessun contenuto di produzione. Questa e la ricostruzione.
Installazione#
pip install fastapi "uvicorn[standard]"L’extra [standard] installa uvloop (event loop piu veloce), httptools (parsing HTTP piu veloce) e websockets. Usalo sempre.
Modelli Pydantic: Request, Response e Validazione#
from datetime import datetime
from typing import Annotated
from pydantic import BaseModel, Field, field_validator
class CreateUserRequest(BaseModel):
username: Annotated[str, Field(min_length=3, max_length=50, pattern=r"^[a-z0-9_]+$")]
email: str
age: Annotated[int, Field(ge=18, le=120)]
@field_validator("email")
@classmethod
def email_must_have_at(cls, v: str) -> str:
if "@" not in v:
raise ValueError("indirizzo email non valido")
return v.lower()
class UserResponse(BaseModel):
id: int
username: str
email: str
created_at: datetime
model_config = {"from_attributes": True} # abilita la modalita ORMPydantic v2 valida alla costruzione. Se age e una stringa che non puo essere convertita in intero, FastAPI restituisce un 422 con un corpo di errore strutturato – nessun codice di validazione manuale richiesto.
Path Parameter, Query Parameter e Request Body#
from fastapi import FastAPI, Query
from models import CreateUserRequest, UserResponse
app = FastAPI(title="User API", version="1.0.0")
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(
user_id: int,
include_deleted: bool = Query(default=False, description="include soft-deleted users"),
):
# user_id e un path parameter (obbligatorio, tipizzato come int)
# include_deleted e un query parameter con un default
user = await db.get_user(user_id, include_deleted=include_deleted)
if user is None:
raise HTTPException(status_code=404, detail="utente non trovato")
return user
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(body: CreateUserRequest):
# body viene automaticamente parsato dal corpo della richiesta JSON
user = await db.create_user(body.username, body.email, body.age)
return userresponse_model=UserResponse e fondamentale. Dice a FastAPI di serializzare il valore restituito attraverso il modello Pydantic, che rimuove i campi non dichiarati in UserResponse. Senza di esso, i campi interni del database (hash delle password, ID interni, flag) possono trapelare ai consumer dell’API.
Endpoint Async: Quando Usare async def vs def#
import httpx
import time
# Usa async def per lavoro I/O-bound: query al database, chiamate HTTP, lettura file
@app.get("/async-example")
async def fetch_remote():
async with httpx.AsyncClient() as client:
resp = await client.get("https://api.example.com/data")
return resp.json()
# Usa def (sync) per lavoro CPU-bound: elaborazione immagini, calcoli pesanti
# FastAPI esegue gli handler sync in un threadpool automaticamente
@app.get("/sync-example")
def compute_heavy():
# Questo viene eseguito in un threadpool, senza bloccare l'event loop
result = expensive_cpu_computation()
return {"result": result}Non chiamare funzioni bloccanti (time.sleep, driver di database sincroni, requests.get) all’interno di handler async def. Bloccano l’event loop e serializzano tutte le richieste. Usa await asyncio.sleep, driver di database async (asyncpg, motor) e httpx.AsyncClient.
Dependency Injection con Depends()#
Le dipendenze sono il modo piu pulito per condividere autenticazione, sessioni di database e configurazione tra gli endpoint.
from fastapi import Depends, HTTPException, Header
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
# --- Dipendenza sessione database ---
async_session_factory: async_sessionmaker[AsyncSession] | None = None
async def get_db() -> AsyncSession:
async with async_session_factory() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
# --- Dipendenza autenticazione JWT Bearer ---
async def get_current_user(
authorization: str = Header(),
db: AsyncSession = Depends(get_db),
) -> dict:
if not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="token bearer mancante")
token = authorization.removeprefix("Bearer ")
payload = verify_jwt(token) # solleva eccezione se il token non e valido
if payload is None:
raise HTTPException(status_code=401, detail="token non valido")
user = await db.get(User, payload["sub"])
if user is None:
raise HTTPException(status_code=401, detail="utente non trovato")
return userfrom fastapi import Depends
from dependencies import get_db, get_current_user
@app.get("/me", response_model=UserResponse)
async def get_profile(current_user=Depends(get_current_user)):
return current_user
@app.delete("/users/{user_id}", status_code=204)
async def delete_user(
user_id: int,
db: AsyncSession = Depends(get_db),
_: dict = Depends(get_current_user), # verifica auth senza usare il valore
):
await db.delete(await db.get(User, user_id))FastAPI risolve automaticamente il grafo delle dipendenze. get_current_user dipende a sua volta da get_db, quindi una singola sessione di database viene condivisa tra entrambe.
HTTPException e Handler di Eccezioni Personalizzati#
from fastapi import Request
from fastapi.responses import JSONResponse
class RateLimitExceeded(Exception):
def __init__(self, retry_after: int):
self.retry_after = retry_after
@app.exception_handler(RateLimitExceeded)
async def rate_limit_handler(request: Request, exc: RateLimitExceeded) -> JSONResponse:
return JSONResponse(
status_code=429,
content={"detail": "limite di frequenza superato", "retry_after": exc.retry_after},
headers={"Retry-After": str(exc.retry_after)},
)Background Task#
Usa BackgroundTasks per lavoro che non deve bloccare la risposta HTTP: invio email, registrazione di analytics, trigger di pipeline async.
from fastapi import BackgroundTasks
async def send_welcome_email(email: str, username: str) -> None:
# viene eseguito dopo l'invio della risposta
await email_client.send(
to=email,
subject="Benvenuto",
body=f"Ciao {username}, il tuo account e pronto.",
)
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(body: CreateUserRequest, background_tasks: BackgroundTasks):
user = await db.create_user(body.username, body.email)
background_tasks.add_task(send_welcome_email, body.email, body.username)
return user # risposta inviata immediatamente; l'email viene dopoBackgroundTasks viene eseguito nello stesso processo dopo l’invio della risposta. Per lavoro in background durevole (retry, persistenza tra i riavvii), usa una coda di task come Celery o ARQ.
Deployment in Produzione#
# Processo singolo, auto-reload al cambio dei file
uvicorn main:app --reload --host 0.0.0.0 --port 8000Non usare mai --reload in produzione. Monitora il filesystem e riavvia il processo – non e sicuro ne performante per il traffico di produzione.
# Worker multipli per parallelismo CPU
# Regola empirica: (2 * numero_CPU) + 1
gunicorn main:app \
--workers 5 \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--timeout 30 \
--graceful-timeout 10 \
--access-logfile -Ogni worker Gunicorn e un processo separato che esegue un event loop Uvicorn. Questo fornisce parallelismo a livello di processo (Gunicorn) e I/O non bloccante (Uvicorn) in combinazione.
Per i deployment containerizzati, usa questo come CMD in un Dockerfile:
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
ENV WORKERS=4
CMD gunicorn main:app \
--workers $WORKERS \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000Carica la configurazione dell’ambiente tramite Pydantic Settings:
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
secret_key: str
environment: str = "production"
debug: bool = False
model_config = {"env_file": ".env"}
settings = Settings()Errori comuni e come evitarli
Usare def invece di async def per handler I/O-bound
Una chiamata al database sincrona all’interno di un handler async def blocca l’event loop mentre attende il database. Ogni altra richiesta si mette in coda. Usa driver async (asyncpg, motor, aioredis) con await.
Non usare response_model
Senza response_model, FastAPI serializza il valore restituito grezzo. Se il tuo handler restituisce un oggetto ORM di SQLAlchemy, i campi interni (password hashate, flag interni, chiavi esterne) appariranno nella risposta. Dichiara sempre response_model su ogni endpoint.
Nessuna dependency injection per le sessioni di database
Creare una nuova connessione al database per ogni handler senza una dipendenza porta a connection leak quando si verificano eccezioni prima che la connessione venga chiusa. La dipendenza get_db con yield garantisce che la sessione venga sempre rollback e chiusa, anche in caso di eccezioni.
Eseguire il server dev Uvicorn in produzione
uvicorn main:app avvia un server a processo singolo senza gestione dei worker. Un’eccezione non gestita abbatte l’intero servizio. Gunicorn gestisce il riavvio dei worker, i reload graceful e la supervisione dei processi.
Firma dell’handler errata
L’handler @app.get("/") non accetta request e response come argomenti posizionali. FastAPI usa l’ispezione della firma della funzione per iniettare path parameter, query parameter, header e body. La firma corretta per un semplice GET senza parametri e async def handler() -> dict.
Se vuoi approfondire questi argomenti, offro sessioni di coaching 1:1 per ingegneri che lavorano su integrazione AI, architettura cloud e platform engineering. Prenota una sessione (50 EUR / 60 min) o scrivimi a manuel.fedele+website@gmail.com.
