Capítulo 17, Intermediário
CRUD com SQLAlchemy
As quatro operações sobre um banco de verdade, com o padrão que eu uso em todo projeto: modelo do banco, modelos do Pydantic para a API, e uma rota enxuta para cada operação.
Dois mundos, dois modelos
O modelo do banco (SQLAlchemy) descreve a tabela. O modelo da API (Pydantic) descreve o que entra e sai. Misturá-los acopla o seu contrato público ao desenho do banco. O from_attributes=True permite ao Pydantic ler diretamente os atributos de um objeto do SQLAlchemy:
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, Response, status
from fastapi.testclient import TestClient
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import String, create_engine, func, select
from sqlalchemy.exc import IntegrityError
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), unique=True)
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)
class TarefaEntrada(BaseModel):
titulo: str = Field(min_length=1, max_length=100)
concluida: bool = False
class TarefaParcial(BaseModel):
titulo: str | None = Field(default=None, min_length=1, max_length=100)
concluida: bool | None = None
class TarefaSaida(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
titulo: str
concluida: bool
def obter_sessao():
with SessionLocal() as sessao:
yield sessao
Sessao = Annotated[Session, Depends(obter_sessao)]
app = FastAPI()
O unique=True no título é de propósito: ele vai nos dar um erro real de banco para tratar.
Criar e ler
O add, o commit e o refresh formam o ciclo de criação. Se o título já existir, o banco recusa, o SQLAlchemy levanta IntegrityError, e a API deve responder 409. Depois de um erro, a sessão precisa de rollback:
def buscar_ou_404(sessao: Session, tarefa_id: int) -> Tarefa:
tarefa = sessao.get(Tarefa, tarefa_id)
if tarefa is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Tarefa não encontrada")
return tarefa
@app.post("/tarefas", response_model=TarefaSaida, status_code=status.HTTP_201_CREATED)
def criar(entrada: TarefaEntrada, sessao: Sessao):
tarefa = Tarefa(**entrada.model_dump())
sessao.add(tarefa)
try:
sessao.commit()
except IntegrityError:
sessao.rollback()
raise HTTPException(status.HTTP_409_CONFLICT, "Já existe uma tarefa com esse título")
sessao.refresh(tarefa)
return tarefa
@app.get("/tarefas", response_model=list[TarefaSaida])
def listar(sessao: Sessao, concluida: bool | None = None):
consulta = select(Tarefa).order_by(Tarefa.id)
if concluida is not None:
consulta = consulta.where(Tarefa.concluida == concluida)
return sessao.scalars(consulta).all()
@app.get("/tarefas/{tarefa_id}", response_model=TarefaSaida)
def obter(tarefa_id: int, sessao: Sessao):
return buscar_ou_404(sessao, tarefa_id)
A lista usa o .where() para filtrar no banco, e não depois, em Python. Com milhares de linhas, é a diferença entre trazer três registros ou trazer tudo.
Atualizar e remover
O PATCH aplica só os campos enviados, com o exclude_unset=True que já conhecemos, e o DELETE devolve 204:
@app.patch("/tarefas/{tarefa_id}", response_model=TarefaSaida)
def alterar(tarefa_id: int, mudancas: TarefaParcial, sessao: Sessao):
tarefa = buscar_ou_404(sessao, tarefa_id)
for campo, valor in mudancas.model_dump(exclude_unset=True).items():
setattr(tarefa, campo, valor)
sessao.commit()
return tarefa
@app.delete("/tarefas/{tarefa_id}", status_code=status.HTTP_204_NO_CONTENT)
def remover(tarefa_id: int, sessao: Sessao):
sessao.delete(buscar_ou_404(sessao, tarefa_id))
sessao.commit()
return Response(status_code=status.HTTP_204_NO_CONTENT)
@app.get("/contagem")
def contagem(sessao: Sessao):
total = sessao.scalar(select(func.count()).select_from(Tarefa))
concluidas = sessao.scalar(select(func.count()).where(Tarefa.concluida))
return {"total": total, "concluidas": concluidas}
O fluxo completo
cliente = TestClient(app)
print(cliente.post("/tarefas", json={"titulo": "Estudar FastAPI"}).json())
print(cliente.post("/tarefas", json={"titulo": "Escrever testes", "concluida": True}).json())
repetida = cliente.post("/tarefas", json={"titulo": "Estudar FastAPI"})
print(repetida.status_code, repetida.json())
print([t["titulo"] for t in cliente.get("/tarefas?concluida=true").json()])
{'id': 1, 'titulo': 'Estudar FastAPI', 'concluida': False}
{'id': 2, 'titulo': 'Escrever testes', 'concluida': True}
409 {'detail': 'Já existe uma tarefa com esse título'}
['Escrever testes']
print(cliente.patch("/tarefas/1", json={"concluida": True}).json())
print(cliente.get("/contagem").json())
print(cliente.delete("/tarefas/2").status_code)
print(cliente.get("/tarefas/2").status_code, cliente.get("/contagem").json())
{'id': 1, 'titulo': 'Estudar FastAPI', 'concluida': True}
{'total': 2, 'concluidas': 2}
204
404 {'total': 1, 'concluidas': 1}
O PATCH mudou só concluida, e o título não foi tocado. A contagem usa func.count(), que faz a conta no banco em vez de carregar as linhas.
Uma sessão por requisição, e nunca esqueça o rollback
Depois de um erro no
commit, a sessão fica em um estado que não aceita novas operações até você chamarrollback(). O padrão é: tente ocommit, no erro façarollbacke responda com o status certo. E a sessão vem da dependênciaobter_sessao: nunca crie uma sessão global.
Exercício 1
Marcar todas como concluídas
Acrescente POST /tarefas/concluir-todas, que marque todas as tarefas pendentes como concluídas com uma única instrução UPDATE (sqlalchemy.update) e devolva quantas foram alteradas.