Chapter 56, Backend

SQLAlchemy

SQLAlchemy maps tables to Python classes. Used with discipline, it eliminates repetitive SQL. Used without discipline, it hides queries that crash it in production.

Code for this chapter: backend/cap56_sqlalchemy_orm.py

Typed models (style 2.0)

Each class is a table, each attribute annotated with Mapped[...] is a column, and mypy understands all of this. The relationship navigate between related tables. To learn and test, use SQLite in memory, and the same code runs on PostgreSQL by just changing the 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)

Session: the unit of work

The Session tracks the objects you loaded and created, and writes everything at once to the commit. I open a session per operation (or per web request) and close it at the end, with 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()

Querying with select

The 2.0 style uses select(...) and the SQL functions from func. What you write is practically the SQL from the previous chapter, just with 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)

Output

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

The N+1 problem

It is the most common performance error with ORM. When iterating over authors and accessing autor.livros, SQLAlchemy, by default, makes a new query for each author (lazy loading). With 1000 authors, there are 1001 queries. The fix is to request the related ones in advance, with selectinload. An event listener counts the queries for you to see the problem:

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")

Output

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

With three authors, the difference is from 4 to 2. With a thousand, it is from 1001 to 2.

Update, delete and transactions

Changing an attribute of a loaded object is enough: in the commit, the session detects the change and issues the UPDATE. The cascade="all, delete-orphan" that we declared makes deleting an author delete their books. And the block with sessao.begin() opens a transaction that confirms itself or rolls back in case of error:

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__)

Output

3
IntegrityError

The second block tries to repeat an author that already exists, and the database refuses due to the restriction UNIQUE. The IntegrityError undoes the entire transaction.

Repository

I isolate the queries in a class of repository, so that the rest of the code does not know SQLAlchemy. It becomes easy to test (with a fake repository) and to change:

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)])

Output

['Quincas Borba', 'Dom Casmurro']

The same model in PostgreSQL

Only the URL changes. With the driver psycopg (version 3), it starts with 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)

Output

postgresql ['caneta']

create_all is not migration
The create_all it only creates tables that do not exist. It does not alter an existing table, does not add a column, does not rename anything. It is useful for tests and for the first day. In production, a migration tool evolves the schema, the subject of the next chapter.

Exercise 1

Titles of an author

Write titulos_do_autor(sessao, nome) that returns, ordered, the titles of the books of an author, with a single query (without N+1).

Typed models (style 2.0)

Session: the unit of work

Querying with select

The N+1 problem

Update, delete and transactions

Repository

The same model in PostgreSQL

View solution