API REST in Python: persistenza con SQLModel

Finora i nostri libri vivevano in un dizionario in memoria, destinato a svanire al riavvio del server. In questa puntata diamo loro una casa stabile: il database. Useremo SQLModel, la libreria che unisce Pydantic e SQLAlchemy in un unico modello, nata dallo stesso autore di FastAPI e pensata proprio per integrarsi con esso. Definiremo il modello Book, configureremo il motore e la sessione, introdurremo una dipendenza che apre e chiude la sessione per ogni richiesta e riscriveremo le rotte sfruttando l'ORM. Vedremo inoltre come gestire l'evoluzione dello schema con le migrazioni di Alembic e come popolare il database con dati di prova.

Installare SQLModel

Con l'ambiente virtuale attivo, installiamo la libreria.

pip install sqlmodel

SQLModel porta con sé SQLAlchemy, il motore ORM sottostante, e si appoggia a Pydantic per la validazione. Un unico modello potrà così fungere sia da tabella del database sia da schema dei dati, anche se, come vedremo nella prossima puntata, per le API conviene separare i due ruoli.

Il modello Book

Creiamo il file app/models.py e vi definiamo il modello. Una classe che eredita da SQLModel e dichiara table=True diventa una tabella; i suoi attributi, annotati con i tipi Python, diventano colonne.

from datetime import datetime, timezone

from sqlmodel import Field, SQLModel


class Book(SQLModel, table=True):
    # id è la chiave primaria: None finché il record non viene salvato
    id: int | None = Field(default=None, primary_key=True)
    title: str
    author: str
    year: int | None = Field(default=None)
    # L'ISBN è unico a livello di database: niente duplicati
    isbn: str | None = Field(default=None, unique=True, index=True)
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

Le annotazioni di tipo fanno un doppio lavoro. Dal punto di vista del database definiscono il tipo di ciascuna colonna; dal punto di vista di Pydantic definiscono come validare i dati. Il tipo int | None per l'identificatore riflette il fatto che prima del salvataggio il libro non ha ancora un id, che verrà assegnato dal database. Il parametro default_factory di created_at calcola l'ora corrente al momento della creazione dell'oggetto.

Configurare il motore e la sessione

Il motore è l'oggetto che gestisce la connessione al database; la sessione è l'unità di lavoro attraverso cui leggiamo e scriviamo. Creiamo il file app/database.py. Per semplicità useremo SQLite, che non richiede l'installazione di alcun server: il database è un semplice file.

from collections.abc import Generator

from sqlmodel import Session, SQLModel, create_engine

DATABASE_URL = "sqlite:///./catalogo.db"

# check_same_thread=False è necessario solo con SQLite in ambiente asincrono
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})


def create_db_and_tables() -> None:
    # Crea tutte le tabelle definite dai modelli SQLModel
    SQLModel.metadata.create_all(engine)


def get_session() -> Generator[Session, None, None]:
    # Apre una sessione per la durata della richiesta e la chiude al termine
    with Session(engine) as session:
        yield session

La funzione get_session è una dipendenza: FastAPI la eseguirà prima di ogni rotta che la richiede, iniettando la sessione come parametro. Il costrutto with combinato con yield garantisce che la sessione venga sempre chiusa al termine della richiesta, anche in caso di errore. Passare a un altro motore, come PostgreSQL, richiederebbe soltanto di cambiare la stringa di connessione, per esempio in postgresql://user:password@localhost/catalogo: il resto del codice resterebbe identico, perché SQLAlchemy astrae le differenze fra i database relazionali.

Per creare le tabelle all'avvio dell'applicazione, colleghiamo create_db_and_tables all'evento di startup nel file app/main.py tramite il gestore del ciclo di vita.

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.database import create_db_and_tables
from app.routers import books, health


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Eseguito all'avvio: crea le tabelle se non esistono
    create_db_and_tables()
    yield
    # Qui potremmo liberare risorse alla chiusura


app = FastAPI(title="Catalogo Libri API", lifespan=lifespan)

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

Creare le tabelle allo startup va benissimo in sviluppo, ma non è il modo giusto di gestire l'evoluzione dello schema in produzione: per quello serve un sistema di migrazioni, che vedremo tra poco.

Una dipendenza per recuperare il libro

Nella puntata precedente ogni rotta su un singolo libro ripeteva la ricerca e il controllo di esistenza. Estraiamo questa logica in una dipendenza dedicata, che recupera il libro e solleva un 404 se non esiste. La collochiamo in app/routers/books.py.

from fastapi import Depends, HTTPException, status
from sqlmodel import Session

from app.database import get_session
from app.models import Book


def get_book_or_404(book_id: int, session: Session = Depends(get_session)) -> Book:
    # Recupera il libro dalla chiave primaria oppure interrompe con un 404
    book = session.get(Book, book_id)
    if book is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Libro non trovato")
    return book

Questa dipendenza ne usa a sua volta un'altra, get_session: FastAPI risolve l'intera catena automaticamente. Ogni rotta che dichiara un parametro book: Book = Depends(get_book_or_404) riceverà il libro già caricato, senza scrivere una riga di ricerca né di gestione dell'errore.

Riscrivere le rotte con l'ORM

Ora possiamo abbandonare l'archivio in memoria e usare il database. Riscriviamo l'intero router, completando tutte le operazioni CRUD. Continuiamo, per un'ultima puntata, ad accettare i dati come dizionario grezzo: nella prossima li tipizzeremo con i modelli Pydantic.

from fastapi import APIRouter, Depends, Request, Response, status
from sqlmodel import Session, select

from app.database import get_session
from app.models import Book

router = APIRouter(prefix="/books", tags=["books"])


@router.get("")
def index(session: Session = Depends(get_session)) -> list[Book]:
    # select(Book) genera l'interrogazione; exec la esegue
    books = session.exec(select(Book).order_by(Book.id.desc())).all()
    return books


@router.post("", status_code=status.HTTP_201_CREATED)
async def store_book(request: Request, session: Session = Depends(get_session)) -> Book:
    data = await request.json()
    book = Book(**data)
    session.add(book)
    session.commit()
    session.refresh(book)  # ricarica il record per ottenere l'id assegnato
    return book


@router.get("/{book_id}")
def show(book: Book = Depends(get_book_or_404)) -> Book:
    # Il libro è già caricato dalla dipendenza
    return book


@router.put("/{book_id}")
async def update_book(
    request: Request,
    book: Book = Depends(get_book_or_404),
    session: Session = Depends(get_session),
) -> Book:
    data = await request.json()
    # Applica solo i campi effettivamente forniti
    for field, value in data.items():
        setattr(book, field, value)
    session.add(book)
    session.commit()
    session.refresh(book)
    return book


@router.delete("/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
def destroy(
    book: Book = Depends(get_book_or_404),
    session: Session = Depends(get_session),
) -> Response:
    session.delete(book)
    session.commit()
    return Response(status_code=status.HTTP_204_NO_CONTENT)

Confrontando con la versione della puntata precedente, la logica di accesso ai dati è ora affidata all'ORM, mentre la ricerca del libro e il controllo di esistenza sono spariti dalle rotte, assorbiti dalla dipendenza. Il ciclo tipico di scrittura è sempre lo stesso: aggiungiamo l'oggetto alla sessione con add, confermiamo con commit e, quando ci serve rileggere i valori generati dal database come l'id, chiamiamo refresh.

Restituire un modello tabella: un problema da risolvere

Restituendo direttamente l'oggetto Book con table=True, FastAPI lo serializza esponendo tutte le sue colonne, comprese quelle che potremmo voler nascondere e nel formato grezzo del database. In un'API pubblica raramente vogliamo questo: preferiamo controllare con precisione quali campi mostrare e in quale forma. Inoltre stiamo costruendo il libro con Book(**data) a partire da dati non validati, il che è pericoloso. Entrambi i problemi hanno una soluzione elegante: il primo lo affronteremo nella prossima puntata separando i modelli di lettura da quelli tabella; il secondo nella puntata sulla validazione, con i modelli di input.

Migrazioni con Alembic

Creare le tabelle con create_all funziona una volta sola: non sa come trasformare uno schema esistente quando aggiungiamo o modifichiamo una colonna. Per gestire l'evoluzione dello schema in modo versionato e riproducibile si usa Alembic, lo strumento di migrazione di SQLAlchemy. Installiamolo e inizializziamolo.

pip install alembic
alembic init migrations

Il comando crea una cartella migrations e un file alembic.ini. Dopo aver configurato la stringa di connessione e collegato i metadati dei modelli SQLModel nel file migrations/env.py, possiamo generare una migrazione automaticamente confrontando i modelli con lo stato del database.

alembic revision --autogenerate -m "crea tabella books"

Alembic ispeziona i modelli e scrive un file di migrazione con le istruzioni per creare la tabella. Applichiamo la migrazione al database.

alembic upgrade head

Se in futuro modificheremo un modello — per esempio aggiungendo la colonna user_id che useremo nella puntata sull'autorizzazione — genereremo una nuova migrazione anziché toccare quella già eseguita: la storia delle modifiche resta così tracciabile e ripetibile su ogni ambiente. In sviluppo continueremo comunque a usare create_all per comodità, riservando Alembic ai casi in cui lo schema deve evolvere in modo controllato.

Popolare il database con dati di prova

Per sviluppare e testare un'API servono dati realistici. Scriviamo un piccolo script di seeding, seed.py, che usa la libreria Faker per generare libri plausibili. Installiamo prima Faker.

pip install faker

from faker import Faker

from app.database import create_db_and_tables, engine
from app.models import Book
from sqlmodel import Session

fake = Faker("it_IT")


def seed(count: int = 20) -> None:
    create_db_and_tables()
    with Session(engine) as session:
        for _ in range(count):
            book = Book(
                title=fake.sentence(nb_words=3),
                author=fake.name(),
                year=fake.random_int(min=1950, max=2025),
                isbn=fake.isbn13(),
            )
            session.add(book)
        session.commit()


if __name__ == "__main__":
    seed()
    print("Database popolato con dati di prova.")

Eseguiamo lo script.

python seed.py

Interrogando ora GET /api/books otterremo l'elenco dei venti libri appena creati. Durante lo sviluppo, cancellare il file catalogo.db e rieseguire lo script è un modo rapidissimo per ripartire da uno stato pulito.

Conclusione

Il nostro catalogo ha finalmente una memoria persistente. Abbiamo definito il modello Book con SQLModel, configurato il motore e una dipendenza che apre la sessione per ogni richiesta, e riscritto tutte le rotte sfruttando l'ORM e una dipendenza di recupero che centralizza il 404. Abbiamo visto come gestire l'evoluzione dello schema con le migrazioni di Alembic e come popolare il database di dati realistici con Faker.

Nella prossima puntata ci concentreremo su come i dati escono dalla nostra API. Separeremo i modelli di lettura dai modelli tabella, useremo il parametro response_model per dare alle risposte una forma pulita e controllata, gestiremo l'inclusione condizionale dei campi e introdurremo la paginazione delle collezioni con i suoi metadati.