SQLAlchemy 2: modelos e sessão

As quatro peças

from sqlalchemy import String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmaker
from sqlalchemy.pool import StaticPool


class Base(DeclarativeBase):
    pass


class Tarefa(Base):
    __tablename__ = "tarefas"

    id: Mapped[int] = mapped_column(primary_key=True)
    titulo: Mapped[str] = mapped_column(String(100))
    concluida: Mapped[bool] = mapped_column(default=False)


engine = create_engine(
    "sqlite://",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)
SessionLocal = sessionmaker(engine, expire_on_commit=False)
Base.metadata.create_all(engine)

Este é o estilo do SQLAlchemy 2: as colunas são declaradas com Mapped[tipo] e mapped_column, e o tipo Python já diz o tipo da coluna. Os tutoriais antigos usam Column(Integer) e declarative_base(), que ainda funcionam mas não têm tipagem.

Para ver o que a classe gerou, dá para imprimir o SQL de criação da tabela:

from sqlalchemy.schema import CreateTable

print(CreateTable(Tarefa.__table__).compile(engine))

CREATE TABLE tarefas (
	id INTEGER NOT NULL, 
	titulo VARCHAR(100) NOT NULL, 
	concluida BOOLEAN NOT NULL, 
	PRIMARY KEY (id)
)

Por que `StaticPool` e `check_same_thread`

Duas linhas do código acima merecem explicação. Um banco SQLite em memória (sqlite://) existe só dentro de uma conexão. Por padrão, cada nova conexão abre outro banco vazio. O StaticPool faz todas compartilharem uma conexão, e o check_same_thread=False permite usá-la de threads diferentes (o TestClient e as rotas def rodam em threads). Isso é uma combinação para testes e demonstrações. Com um arquivo de banco ou com o PostgreSQL, você usa o pool padrão e remove as duas opções.

A sessão: adicionar, ler, alterar

with SessionLocal() as sessao:
    sessao.add(Tarefa(titulo="Estudar SQLAlchemy"))
    sessao.add(Tarefa(titulo="Escrever testes", concluida=True))
    sessao.commit()

with SessionLocal() as sessao:
    todas = sessao.scalars(select(Tarefa).order_by(Tarefa.id)).all()
    print([(t.id, t.titulo, t.concluida) for t in todas])

    primeira = sessao.get(Tarefa, 1)
    primeira.concluida = True
    sessao.commit()

with SessionLocal() as sessao:
    print(sessao.get(Tarefa, 1).concluida, sessao.get(Tarefa, 99))

[(1, 'Estudar SQLAlchemy', False), (2, 'Escrever testes', True)]
True None

O add apenas marca o objeto, e é o commit que grava. O sessao.get(Modelo, id) busca pela chave primária (e devolve None se não existe). O select(...) monta consultas mais gerais, e o scalars(...).all() devolve os objetos.

`commit`, `refresh` e `expire_on_commit`

Por padrão, depois de um commit o SQLAlchemy marca os objetos como expirados, para o próximo acesso recarregar os valores do banco. Isso é seguro, mas tem um efeito colateral em APIs: se a sessão já estiver fechada quando a resposta for montada, acessar um atributo expirado falha:

from sqlalchemy.orm.exc import DetachedInstanceError

padrao = sessionmaker(engine)

with padrao() as s:
    t = Tarefa(titulo="Expira")
    s.add(t)
    s.commit()
try:
    print(t.titulo)
except DetachedInstanceError:
    print("DetachedInstanceError: o objeto expirou e a sessão já fechou")

with SessionLocal() as s:
    t2 = Tarefa(titulo="Não expira")
    s.add(t2)
    s.commit()
print(t2.titulo)

DetachedInstanceError: o objeto expirou e a sessão já fechou
Não expira

É esse o motivo pelo qual os tutoriais mandam chamar db.refresh(objeto). Com expire_on_commit=False (como configurei), os objetos mantêm os valores depois do commit, e o refresh só é necessário quando o banco gera um valor que você quer ler (um carimbo de data de criação, por exemplo).

Uma sessão por requisição

A sessão não deve ser global nem compartilhada: cada requisição abre a sua e a fecha no fim. Isso se escreve como uma dependência com yield, como no capítulo 13:

from typing import Annotated

from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient

app = FastAPI()


def obter_sessao():
    with SessionLocal() as sessao:
        yield sessao


Sessao = Annotated[Session, Depends(obter_sessao)]


@app.get("/tarefas/contagem")
def contar(sessao: Sessao):
    return {"total": len(sessao.scalars(select(Tarefa)).all())}


print(TestClient(app).get("/tarefas/contagem").json())

{'total': 4}

E as migrações?

O Base.metadata.create_all(engine) cria as tabelas que não existem, mas não altera as existentes. Para evoluir o esquema de um banco real (acrescentar uma coluna sem perder dados), use o Alembic, como mostrei no capítulo 57 do curso de Python. O create_all serve para aprender e para os testes.