Capítulo 56, Backend

SQLAlchemy

O SQLAlchemy mapeia tabelas para classes Python. Usado com disciplina, ele elimina SQL repetitivo. Usado sem disciplina, ele esconde consultas que o derrubam em produção.

Código deste capítulo: backend/cap56_sqlalchemy_orm.py

Modelos tipados (estilo 2.0)

Cada classe é uma tabela, cada atributo anotado com Mapped[...] é uma coluna, e o mypy entende tudo isso. As relationship navegam entre tabelas relacionadas. Para aprender e testar uso o SQLite em memória, e o mesmo código roda no PostgreSQL trocando só a URL:

backend/cap56_sqlalchemy_orm.pylinhas 10 a 46

from sqlalchemy import ForeignKey, String, create_engine, func, select
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    Session,
    mapped_column,
    relationship,
    selectinload,
)


class Base(DeclarativeBase):
    pass


class Autor(Base):
    __tablename__ = "autores"

    id: Mapped[int] = mapped_column(primary_key=True)
    nome: Mapped[str] = mapped_column(String(80), unique=True)
    livros: Mapped[list["Livro"]] = relationship(
        back_populates="autor", cascade="all, delete-orphan"
    )


class Livro(Base):
    __tablename__ = "livros"

    id: Mapped[int] = mapped_column(primary_key=True)
    titulo: Mapped[str] = mapped_column(String(120))
    paginas: Mapped[int]
    autor_id: Mapped[int] = mapped_column(ForeignKey("autores.id"))
    autor: Mapped[Autor] = relationship(back_populates="livros")


engine = create_engine("sqlite://")
Base.metadata.create_all(engine)

Sessão: a unidade de trabalho

A Session acompanha os objetos que você carregou e criou, e grava tudo de uma vez no commit. Eu abro uma sessão por operação (ou por requisição web) e a fecho no fim, com with:

backend/cap56_sqlalchemy_orm.pylinhas 51 a 60

with Session(engine) as sessao:
    sessao.add_all([
        Autor(nome="Machado de Assis", livros=[
            Livro(titulo="Dom Casmurro", paginas=256),
            Livro(titulo="Quincas Borba", paginas=300),
        ]),
        Autor(nome="Clarice Lispector", livros=[Livro(titulo="A Hora da Estrela", paginas=88)]),
        Autor(nome="Graciliano Ramos", livros=[Livro(titulo="Vidas Secas", paginas=176)]),
    ])
    sessao.commit()

Consultar com select

O estilo 2.0 usa select(...) e as funções de SQL de func. O que você escreve é praticamente o SQL do capítulo anterior, só que com classes:

backend/cap56_sqlalchemy_orm.pylinhas 65 a 72

with Session(engine) as sessao:
    consulta = select(Livro).where(Livro.paginas > 100).order_by(Livro.titulo)
    print([livro.titulo for livro in sessao.scalars(consulta)])
    print(sessao.scalar(select(func.sum(Livro.paginas))))
    por_autor = sessao.execute(
        select(Autor.nome, func.count(Livro.id)).join(Livro).group_by(Autor.nome).order_by(Autor.nome)
    ).all()
    print(por_autor)

Saída

['Dom Casmurro', 'Quincas Borba', 'Vidas Secas']
820
[('Clarice Lispector', 1), ('Graciliano Ramos', 1), ('Machado de Assis', 2)]

O problema N+1

É o erro de desempenho mais comum com ORM. Ao percorrer os autores e acessar autor.livros, o SQLAlchemy, por padrão, faz uma consulta nova para cada autor (carregamento preguiçoso). Com 1000 autores, são 1001 consultas. A correção é pedir os relacionados de antemão, com selectinload. Um ouvinte de eventos conta as consultas para você enxergar o problema:

backend/cap56_sqlalchemy_orm.pylinhas 77 a 96

from sqlalchemy import event

consultas = []
event.listen(
    engine,
    "before_cursor_execute",
    lambda conexao, cursor, comando, parametros, contexto, varias: consultas.append(comando),
)

with Session(engine) as sessao:
    consultas.clear()
    for autor in sessao.scalars(select(Autor)):
        len(autor.livros)
    print("carregamento preguiçoso:", len(consultas), "consultas")

with Session(engine) as sessao:
    consultas.clear()
    for autor in sessao.scalars(select(Autor).options(selectinload(Autor.livros))):
        len(autor.livros)
    print("com selectinload:", len(consultas), "consultas")

Saída

carregamento preguiçoso: 4 consultas
com selectinload: 2 consultas

Com três autores a diferença é de 4 para 2. Com mil, é de 1001 para 2.

Alterar, apagar e transações

Alterar um atributo de um objeto carregado basta: no commit, a sessão detecta a mudança e emite o UPDATE. O cascade="all, delete-orphan" que declaramos faz apagar um autor apagar os seus livros. E o bloco with sessao.begin() abre uma transação que confirma sozinha ou desfaz em caso de erro:

backend/cap56_sqlalchemy_orm.pylinhas 101 a 114

with Session(engine) as sessao:
    livro = sessao.scalars(select(Livro).where(Livro.titulo == "Vidas Secas")).one()
    livro.paginas = 180
    sessao.commit()
    autor = sessao.scalars(select(Autor).where(Autor.nome == "Graciliano Ramos")).one()
    sessao.delete(autor)
    sessao.commit()
    print(sessao.scalar(select(func.count(Livro.id))))

try:
    with Session(engine) as sessao, sessao.begin():
        sessao.add(Autor(nome="Machado de Assis"))
except Exception as erro:
    print(type(erro).__name__)

Saída

3
IntegrityError

O segundo bloco tenta repetir um autor que já existe, e o banco recusa pela restrição UNIQUE. O IntegrityError desfaz a transação inteira.

Repositório

Eu isolo as consultas em uma classe de repositório, para que o resto do código não conheça o SQLAlchemy. Fica fácil de testar (com um repositório falso) e de trocar:

backend/cap56_sqlalchemy_orm.pylinhas 119 a 133

class RepositorioLivros:
    def __init__(self, sessao: Session) -> None:
        self._sessao = sessao

    def por_titulo(self, titulo: str) -> Livro | None:
        return self._sessao.scalars(select(Livro).where(Livro.titulo == titulo)).first()

    def mais_longos(self, quantidade: int) -> list[Livro]:
        consulta = select(Livro).order_by(Livro.paginas.desc()).limit(quantidade)
        return list(self._sessao.scalars(consulta))


with Session(engine) as sessao:
    repositorio = RepositorioLivros(sessao)
    print([livro.titulo for livro in repositorio.mais_longos(2)])

Saída

['Quincas Borba', 'Dom Casmurro']

O mesmo modelo no PostgreSQL

Só a URL muda. Com o driver psycopg (versão 3), ela começa com postgresql+psycopg://:

O mesmo código no PostgreSQL

import os

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


class Base(DeclarativeBase):
    pass


class Produto(Base):
    __tablename__ = "produtos_sa"

    id: Mapped[int] = mapped_column(primary_key=True)
    nome: Mapped[str] = mapped_column(String(80))


url = os.environ["DATABASE_URL"].replace("postgresql://", "postgresql+psycopg://", 1)
engine = create_engine(url)
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
with Session(engine) as sessao:
    sessao.add(Produto(nome="caneta"))
    sessao.commit()
    print(engine.dialect.name, sessao.scalars(select(Produto.nome)).all())
Base.metadata.drop_all(engine)

Saída

postgresql ['caneta']

create_all não é migração
O create_all só cria tabelas que não existem. Ele não altera uma tabela existente, não adiciona coluna, não renomeia nada. Serve para testes e para o primeiro dia. Em produção, quem evolui o esquema é uma ferramenta de migração, o assunto do próximo capítulo.

Exercício 1

Títulos de um autor

Escreva titulos_do_autor(sessao, nome) que devolva, ordenados, os títulos dos livros de um autor, com uma única consulta (sem N+1).

Modelos tipados (estilo 2.0)

Sessão: a unidade de trabalho

Consultar com select

O problema N+1

Alterar, apagar e transações

Repositório

O mesmo modelo no PostgreSQL

Ver solução