API REST in Python: errori, versioning, rate limiting e test

Concludiamo la serie mettendo insieme i tasselli che distinguono un'API funzionante da un'API pronta per la produzione. Renderemo uniforme la gestione degli errori adottando lo standard problem+json, adotteremo una strategia di versioning per far evolvere l'interfaccia senza rompere i client esistenti, proteggeremo gli endpoint dall'abuso con il rate limiting e, soprattutto, scriveremo test automatici che verifichino il comportamento dell'intera API. Al termine il nostro catalogo di libri sarà un servizio completo e affidabile.

Gestione uniforme degli errori

Un'API dovrebbe rispondere agli errori sempre nello stesso formato, qualunque ne sia la causa. In FastAPI gli errori vengono gestiti dagli exception handler, funzioni che intercettano un tipo di eccezione e ne producono la risposta. Fino a qui abbiamo usato HTTPException, che genera un corpo con la sola chiave detail, mentre gli errori di validazione producono la struttura con detail come lista. Due formati diversi: uniformiamoli.

Lo standard problem+json

Esiste uno standard formale per rappresentare gli errori nelle API HTTP: la RFC 9457, che definisce il tipo di media application/problem+json. Un corpo di errore conforme contiene alcuni campi convenzionali: type, un URI che identifica il tipo di problema; title, un riassunto leggibile; status, il codice HTTP; detail, una spiegazione specifica dell'occorrenza. Adottare questo formato rende gli errori prevedibili e interoperabili. Registriamo degli handler in app/main.py per convertire le eccezioni più comuni in questo formato.

from fastapi import FastAPI, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException


def problem_response(status_code: int, title: str, detail: str, extra: dict | None = None) -> JSONResponse:
    body = {
        "type": f"https://httpstatuses.io/{status_code}",
        "title": title,
        "status": status_code,
        "detail": detail,
    }
    if extra:
        body.update(extra)
    # Il tipo di media problem+json segnala che il corpo è un errore strutturato
    return JSONResponse(status_code=status_code, content=body, media_type="application/problem+json")


@app.exception_handler(StarletteHTTPException)
def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return problem_response(
        status_code=exc.status_code,
        title="Errore nella richiesta",
        detail=str(exc.detail),
    )


@app.exception_handler(RequestValidationError)
def validation_exception_handler(request: Request, exc: RequestValidationError):
    return problem_response(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        title="Dati non validi",
        detail="Uno o più campi non hanno superato la validazione.",
        extra={"errors": exc.errors()},  # conserva il dettaglio campo per campo
    )

Con questi due handler, sia un 404 sollevato dal recupero di un libro sia un errore di validazione producono ora un corpo con la stessa struttura di base. Per gli errori di validazione conserviamo il dettaglio campo per campo nella chiave errors, così da non perdere l'informazione preziosa che FastAPI raccoglie. Possiamo aggiungere handler analoghi per il 401 e il 403, costruendo un vocabolario di errori uniforme in tutta l'API.

Eccezioni di dominio personalizzate

Per gli errori specifici del nostro dominio conviene definire eccezioni dedicate e un handler che le renda in formato problem+json. Creiamone una per il caso in cui si tenti di eliminare un libro ancora in prestito.

class BookOnLoanError(Exception):
    def __init__(self, book_id: int) -> None:
        self.book_id = book_id


@app.exception_handler(BookOnLoanError)
def book_on_loan_handler(request: Request, exc: BookOnLoanError):
    return problem_response(
        status_code=status.HTTP_409_CONFLICT,
        title="Libro in prestito",
        detail=f"Impossibile eliminare il libro {exc.book_id}: risulta in prestito.",
    )

Nella rotta ci limitiamo a sollevare l'eccezione quando la condizione si verifica; alla resa penserà l'handler registrato.

@router.delete("/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def destroy(book: Book = Depends(get_owned_book), session: Session = Depends(get_session)):
    if book.is_on_loan():
        raise BookOnLoanError(book_id=book.id)
    session.delete(book)
    session.commit()
    return Response(status_code=status.HTTP_204_NO_CONTENT)

Versioning dell'API

Un'API pubblica evolve, e le modifiche incompatibili vanno introdotte senza rompere i client esistenti. La strategia più diffusa è il versioning tramite prefisso nell'URL. Poiché già organizziamo le rotte in router, aggiungere una versione è naturale: raccogliamo i router di ciascuna versione sotto un prefisso dedicato, così da poter far convivere una v1 stabile e una v2 in evoluzione.

from fastapi import APIRouter

from app.routers import auth, health
from app.routers import books as books_v1
from app.routers.v2 import books as books_v2

# Router di versione 1
v1 = APIRouter(prefix="/v1")
v1.include_router(books_v1.router)
v1.include_router(auth.router)

# Router di versione 2, con la nuova forma delle risorse
v2 = APIRouter(prefix="/v2")
v2.include_router(books_v2.router)

app.include_router(health.router, prefix="/api")
app.include_router(v1, prefix="/api")
app.include_router(v2, prefix="/api")

L'elenco dei libri della prima versione diventa così GET /api/v1/books. Organizzare i router e i modelli di lettura in sottopacchetti per versione, come app/routers/v2, mantiene il codice ordinato man mano che le versioni si accumulano: una v1 e una v2 possono esporre forme diverse dello stesso modello, isolando i cambiamenti di contratto senza intaccare i client già in produzione.

Rate limiting

Per proteggere l'API da un uso eccessivo, accidentale o malevolo, applichiamo dei limiti di frequenza. Una libreria consolidata per FastAPI è slowapi. Installiamola.

pip install slowapi

Configuriamo un limitatore che identifica il chiamante in base al suo indirizzo IP, lo registriamo nell'applicazione e colleghiamo l'handler che risponde quando il limite viene superato.

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address

# Identifica il client dall'indirizzo IP
limiter = Limiter(key_func=get_remote_address)

app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

Applichiamo poi il limite a una rotta con un decoratore, indicando la soglia. La path operation deve ricevere il parametro request, da cui il limitatore ricava l'IP.

from fastapi import Request


@router.post("", response_model=BookRead, status_code=status.HTTP_201_CREATED)
@limiter.limit("30/minute")  # al più 30 creazioni al minuto per IP
def store_book(
    request: Request,
    data: BookCreate,
    session: Session = Depends(get_session),
    user: User = Depends(get_current_user),
):
    book = Book.model_validate(data)
    book.user_id = user.id
    session.add(book)
    session.commit()
    session.refresh(book)
    return book

Quando un client supera il limite riceve una risposta 429 Too Many Requests, corredata dalle intestazioni che gli comunicano quante richieste gli restano e quando potrà riprovare. Un client ben educato le legge per regolare la propria frequenza di chiamata.

Testare l'API

Arriviamo al tassello più importante per la manutenibilità: i test. FastAPI mette a disposizione TestClient, che effettua vere richieste HTTP all'applicazione in memoria, senza bisogno di avviare un server. Lo useremo insieme a pytest, il runner di test standard in Python. Installiamoli.

pip install pytest httpx

Il punto delicato è isolare i test dal database di sviluppo: ogni test deve partire da uno stato pulito. Sfruttiamo il meccanismo delle dipendenze, sostituendo get_session con una versione che punta a un database SQLite in memoria, ricreato per ogni test. Predisponiamo le fixture in un file tests/conftest.py.

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool

from app.database import get_session
from app.main import app


@pytest.fixture(name="session")
def session_fixture():
    # Database in memoria, isolato e ricreato per ogni test
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session


@pytest.fixture(name="client")
def client_fixture(session):
    # Sostituisce la dipendenza della sessione con quella di test
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()

La sostituzione tramite dependency_overrides è la chiave: la stessa dipendenza get_session che in produzione apre il database reale, nei test punta a quello in memoria, senza che il codice dell'applicazione ne sappia nulla. Scriviamo ora un primo test per la lettura pubblica dell'elenco, in tests/test_books.py.

from app.models import Book


def test_l_elenco_dei_libri_e_pubblico(client, session):
    # Prepariamo tre libri nel database di test
    for i in range(3):
        session.add(Book(title=f"Libro {i}", author="Autore"))
    session.commit()

    response = client.get("/api/v1/books")

    assert response.status_code == 200
    body = response.json()
    assert body["meta"]["total"] == 3
    assert len(body["data"]) == 3

Aggiungiamo un test che verifica la validazione: una richiesta di creazione senza titolo deve fallire con codice 422. Poiché la creazione richiede l'autenticazione, prima registriamo un utente e ne usiamo il token.

def register_and_token(client) -> str:
    # Utility: registra un utente di test e restituisce il suo token
    response = client.post("/api/v1/auth/register", json={
        "name": "Tester",
        "email": "tester@example.com",
        "password": "password123",
    })
    return response.json()["access_token"]


def test_la_creazione_richiede_il_titolo(client):
    token = register_and_token(client)
    headers = {"Authorization": f"Bearer {token}"}

    response = client.post("/api/v1/books", json={"author": "Anonimo"}, headers=headers)

    assert response.status_code == 422

Scriviamo ora un test del percorso felice per la creazione, verificando sia la risposta sia la presenza del record nel database.

def test_un_utente_autenticato_crea_un_libro(client, session):
    token = register_and_token(client)
    headers = {"Authorization": f"Bearer {token}"}

    payload = {"title": "Il deserto dei Tartari", "author": "Dino Buzzati", "year": 1940}
    response = client.post("/api/v1/books", json=payload, headers=headers)

    assert response.status_code == 201
    assert response.json()["title"] == "Il deserto dei Tartari"

    # Il libro deve esistere davvero nel database
    books = session.query(Book).all()
    assert len(books) == 1

Infine verifichiamo l'autorizzazione: un utente non deve poter eliminare il libro di un altro. Questo test difende una regola di dominio importante, e il fatto di poterlo esprimere con poche righe è la ricompensa di aver strutturato bene il codice nelle puntate precedenti.

def test_un_utente_non_puo_eliminare_il_libro_altrui(client, session):
    # Primo utente: crea un libro
    token_a = register_and_token(client)
    headers_a = {"Authorization": f"Bearer {token_a}"}
    created = client.post(
        "/api/v1/books",
        json={"title": "Le città invisibili", "author": "Italo Calvino"},
        headers=headers_a,
    ).json()

    # Secondo utente: tenta di eliminarlo
    second = client.post("/api/v1/auth/register", json={
        "name": "Altro", "email": "altro@example.com", "password": "password123",
    })
    headers_b = {"Authorization": f"Bearer {second.json()['access_token']}"}

    response = client.delete(f"/api/v1/books/{created['id']}", headers=headers_b)

    # Deve ricevere un 403 e il libro deve restare nel database
    assert response.status_code == 403
    assert session.get(Book, created["id"]) is not None

Eseguiamo l'intera suite.

pytest

Ogni test verde è una garanzia che una parte del comportamento dell'API continua a funzionare. Quando in futuro modificheremo il codice, la suite ci avviserà immediatamente se abbiamo rotto qualcosa: è la rete di sicurezza che rende serena l'evoluzione di un progetto.

Conclusione della serie

Con questa settima puntata il nostro catalogo di libri è diventato un'API REST completa. Nel corso della serie siamo partiti dai fondamenti di REST e da un primo endpoint, abbiamo costruito le rotte di risorsa con i router, dato persistenza ai dati con SQLModel, plasmato le risposte con i modelli e response_model, difeso gli ingressi con la validazione di Pydantic, messo in sicurezza tutto con l'autenticazione JWT e l'autorizzazione basata sulla proprietà, e infine reso il servizio robusto con la gestione uniforme degli errori, il versioning, il rate limiting e una suite di test.

Il percorso ha seguito una progressione deliberata: ogni puntata ha aggiunto uno strato allo stesso progetto, mostrando come le funzionalità di FastAPI e del suo ecosistema si compongano fra loro in modo naturale, con le dipendenze a fare da collante. Da qui puoi proseguire in autonomia esplorando temi avanzati come l'esecuzione asincrona con un database async, il caching delle risposte, l'elaborazione in background con le code o l'osservabilità in produzione. Le fondamenta, però, sono ormai solide: hai un'API RESTful ben strutturata, sicura e verificata dai test.