Chapter 17, Intermediate
CRUD with SQLAlchemy
The four operations on a real database, with the pattern I use in every project: database model, Pydantic models for the API, and a lean route for each operation.
Two worlds, two models
The model of the database (SQLAlchemy) describes the table. The model of the API (Pydantic) describes what goes in and out. Mixing them couples your public contract to the database design. The from_attributes=True allows Pydantic to read directly the attributes of a SQLAlchemy object:
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()
The unique=True the title is intentional: it will give us a real database error to handle.
Create and read
The add, the commit and the refresh form the creation cycle. If the title already exists, the database rejects it, SQLAlchemy raises IntegrityError, and the API should respond 409. After an error, the session needs 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)
The list uses the .where() to filter in the database, and not later, in Python. With thousands of rows, it's the difference between bringing three records or bringing everything.
Update and remove
The PATCH applies only the fields sent, with the exclude_unset=True we already know, and the DELETE returns 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}
The complete flow
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}
The PATCH changed only concluida, and the title was not touched. The count uses func.count(), which does the count in the database instead of loading the rows.
One session per request, and never forget the rollback
After an error in the
commit, the session remains in a state that does not accept new operations until you callrollback(). The pattern is: try thecommit, on error dorollbackand respond with the correct status. And the session comes from the dependencyobter_sessao: never create a global session.
Exercise 1
Mark all as completed
Add POST /tarefas/concluir-todas, which marks all pending tasks as completed with a single statement UPDATE (sqlalchemy.update) and return how many were changed.