SQLAlchemy 2: models and session

The four pieces

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)

This is the style of SQLAlchemy 2: columns are declared with Mapped[tipo] and mapped_column, and the Python type already indicates the column type. The old tutorials use Column(Integer) and declarative_base(), which still work but do not have typing.

To see what the class generated, you can print the SQL for creating the table:

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

Why `StaticPool` and `check_same_thread`

Two lines of the code above deserve explanation. An SQLite database in memory (sqlite://) exists only within a connection. By default, each new connection opens another empty database. The StaticPool makes all share a connection, and the check_same_thread=False allows it to be used from different threads (the TestClient and the routes def run on threads). This is a combination for tests and demonstrations. With a database file or with PostgreSQL, you use the default pool and remove the two options.

The session: add, read, update

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

The add just marks the object, and it is the commit that writes. The sessao.get(Modelo, id) searches by the primary key (and returns None if it does not exist). The select(...) builds more general queries, and the scalars(...).all() returns the objects.

`commit`, `refresh` and `expire_on_commit`

By default, after a commit SQLAlchemy marks the objects as expired, for the next access to reload the values from the database. This is safe, but has a side effect in APIs: if the session is already closed when the response is built, accessing an expired attribute fails:

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

This is the reason why tutorials instruct to call db.refresh(objeto). With expire_on_commit=False (as I configured), the objects retain the values after the commit, and the refresh is only necessary when the database generates a value that you want to read (a creation timestamp, for example).

One session per request

The session should not be global or shared: each request opens its own and closes it at the end. This is written as a dependency with yield, as in chapter 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}

And the migrations?

The Base.metadata.create_all(engine) creates the tables that do not exist, but does not alter the existing ones. To evolve the schema of a real database (add a column without losing data), use the Alembic, as I showed in chapter 57 of the Python course. The create_all is for learning and for testing.