Guia Prático: SQLAlchemy 2.0

09 Oct, 2026

O SQLAlchemy é o kit de ferramentas SQL e Object-Relational Mapping (ORM) mais adotado no ecossistema Python. Ele oferece flexibilidade em duas camadas principais:

SQLAlchemy Core: Camada orientada a esquema e abstração SQL (gerenciamento de conexões, pools, montagem de queries SQL idiomáticas).
SQLAlchemy ORM: Camada de alto nível que mapeia classes Python para tabelas de banco de dados relacionais e gerencia o ciclo de vida dos objetos por meio de sessões (Unit of Work).

Nota: Este guia utiliza o padrão SQLAlchemy 2.0+, que introduziu tipagem estática integrada (mapped_column, Mapped), consultas com a instrução select() no ORM e uma sintaxe mais coesa e moderna.

1. Instalação e Preparação

Para começar, instale o SQLAlchemy. Utilizaremos o SQLite (já embutido no Python) para os exemplos práticos, mas o código é facilmente portável para PostgreSQL, MySQL ou Oracle:

pip install sqlalchemy

(Se for usar PostgreSQL, instale também o driver psycopg ou psycopg2-binary).

2. Componentes Fundamentais

Antes de escrever código, é importante entender os três pilares do SQLAlchemy:

Engine: O ponto de entrada da conexão. Gerencia o pool de conexões com o banco de dados.
Declarative Base: A classe base a partir da qual seus modelos/tabelas herdam.
Session: O intermediário que rastreia alterações nos objetos e executa comandos atômicos no banco de dados.

3. Configurando a Conexão (`Engine`)

O create_engine inicializa a comunicação com a base:

from sqlalchemy import create_engine

# SQLite em arquivo local:
# engine = create_engine("sqlite:///meu_banco.db", echo=True)

# SQLite volátil em memória (ideal para testes rápidos):
engine = create_engine("sqlite:///:memory:", echo=False)

# Para PostgreSQL (exemplo):
# engine = create_engine("postgresql+psycopg://usuario:senha@localhost:5432/nomedobanco")

Dica: O parâmetro echo=True imprime todas as queries SQL geradas no terminal, sendo muito útil para depuração.

4. Definindo Modelos Declarativos (Sintaxe 2.0)

Na versão 2.0, utilizamos Mapped e mapped_column para obter autocomplete de tipos e checagem estática (compatível com mypy e IDEs).

Vamos criar dois modelos relacionados: Usuario e Tarefa.

from datetime import datetime
from typing import List, Optional
from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


# 1. Classe base declarativa
class Base(DeclarativeBase):
    pass


# 2. Modelo Usuario
class Usuario(Base):
    __tablename__ = "usuarios"

    id: Mapped[int] = mapped_column(primary_key=True)
    nome: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100), unique=True, index=True)
    ativo: Mapped[bool] = mapped_column(default=True)

    # Relacionamento 1 para N (Um usuário tem várias tarefas)
    tarefas: Mapped[List["Tarefa"]] = relationship(
        back_populates="autor", cascade="all, delete-orphan"
    )

    def __repr__(self) -> str:
        return f"Usuario(id={self.id!r}, nome={self.nome!r}, email={self.email!r})"


# 3. Modelo Tarefa
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)
    criado_em: Mapped[datetime] = mapped_column(default=datetime.utcnow)

    # Chave estrangeira
    usuario_id: Mapped[int] = mapped_column(ForeignKey("usuarios.id"))

    # Referência de volta para o Usuario
    autor: Mapped["Usuario"] = relationship(back_populates="tarefas")

    def __repr__(self) -> str:
        return f"Tarefa(id={self.id!r}, titulo={self.titulo!r}, concluida={self.concluida!r})"

5. Criando as Tabelas no Banco

Para gerar fisicamente o esquema definido no banco:

# Cria todas as tabelas registradas na Base
Base.metadata.create_all(engine)

6. Operações de CRUD com `Session`

No SQLAlchemy 2.0, a Session deve ser utilizada preferencialmente com gerenciador de contexto (with). O bloco Session.begin() garante abertura e commit automático (com rollback em caso de exceção).

from sqlalchemy import select
from sqlalchemy.orm import Session

6.1. Create (Inserção)

with Session(engine) as session:
    with session.begin():
        # Criando usuário com tarefas associadas
        novo_usuario = Usuario(
            nome="Wilber Godoy",
            email="wilber@exemplo.com",
            tarefas=[
                Tarefa(titulo="Aprender SQLAlchemy 2.0"),
                Tarefa(titulo="Estruturar pipeline de automação"),
            ],
        )

        outro_usuario = Usuario(nome="Carlos Souza", email="carlos@exemplo.com")

        # session.add() para um item, session.add_all() para lista
        session.add_all([novo_usuario, outro_usuario])

# Ao sair do bloco 'session.begin()', o commit é realizado automaticamente!

6.2. Read (Consultas com `select`)

No estilo 2.0, não se usa mais session.query(Usuario). Usa-se a instrução select(Usuario):

with Session(engine) as session:
    # 1. Buscar todos os usuários
    stmt = select(Usuario).order_by(Usuario.nome)
    usuarios = session.scalars(stmt).all()
    print("--- Todos os usuários ---")
    for u in usuarios:
        print(u)

    # 2. Filtrar por condição (WHERE)
    stmt_filtro = select(Usuario).where(Usuario.email == "wilber@exemplo.com")
    usuario_encontrado = session.scalars(stmt_filtro).first()
    print("\n--- Usuário por e-mail ---")
    print(usuario_encontrado)

    # 3. Buscar pela chave primária (forma direta e otimizada)
    usuario_por_id = session.get(Usuario, 1)
    print("\n--- Busca direta por ID ---")
    print(usuario_por_id)

Por que usar session.scalars()? A query select(Usuario) retorna tuplas de linhas por padrão ((Usuario, )). O método .scalars() desempacota o primeiro elemento de cada tupla, entregando os objetos Usuario diretamente.

6.3. Update (Atualização)

O SQLAlchemy rastreia o estado das instâncias na memória. Basta alterar atributos do objeto carregado dentro de uma transação:

with Session(engine) as session:
    with session.begin():
        # Busca o registro
        tarefa = session.get(Tarefa, 1)
        if tarefa:
            # Modifica diretamente os atributos
            tarefa.concluida = True
            tarefa.titulo = "Aprender SQLAlchemy 2.0 (Concluído!)"

# O SQLAlchemy detecta as alterações (dirty tracking) e emite o UPDATE no commit.

Também é possível fazer atualizações em lote via comando update():

from sqlalchemy import update

with Session(engine) as session:
    with session.begin():
        stmt_update = (
            update(Tarefa)
            .where(Tarefa.concluida == False)
            .values(titulo="Pendente de revisão")
        )
        session.execute(stmt_update)

6.4. Delete (Exclusão)

with Session(engine) as session:
    with session.begin():
        usuario_para_remover = session.get(Usuario, 2)
        if usuario_para_remover:
            session.delete(usuario_para_remover)
            print("Usuário removido com sucesso!")

7. Consultas Avançadas: Joins e Filtros Múltiplos

Realizando `JOIN` entre tabelas:

with Session(engine) as session:
    # Seleciona títulos de tarefas de usuários ativos
    stmt = (
        select(Tarefa.titulo, Usuario.nome)
        .join(Tarefa.autor)
        .where(Usuario.ativo == True)
        .where(Tarefa.concluida == True)
    )

    resultados = session.execute(stmt).all()
    for titulo, autor_nome in resultados:
        print(f"Tarefa: '{titulo}' | Autor: {autor_nome}")

Carregamento Otimizado de Relacionamentos (`selectinload`):

Por padrão, relacionamentos usam Lazy Loading (carregam os dados sob demanda quando o atributo é acessado), o que pode disparar o problema de N+1 queries. Para evitar isso, use selectinload:

from sqlalchemy.orm import selectinload

with Session(engine) as session:
    stmt = select(Usuario).options(selectinload(Usuario.tarefas))
    usuarios_com_tarefas = session.scalars(stmt).all()

    for u in usuarios_com_tarefas:
        print(f"Usuário: {u.nome} tem {len(u.tarefas)} tarefas.")

8. Boas Práticas em Projetos Reais

Um Engine por aplicação: Crie a instância do engine uma única vez no escopo global ou de configuração do projeto. O pool de conexões é gerenciado internamente por ele.
Sessões de escopo curto: Não mantenha a Session aberta por longos períodos. Em APIs (FastAPI/Flask) ou rotinas automatizadas, abra a sessão, execute a operação, dê commit/rollback e feche-a imediatamente.
Use o padrão Repository para desacoplar regras: Evite espalhar chamadas de Session por todo o código de negócio; isole a lógica de persistência em classes ou módulos dedicados.
Migrações de Banco com Alembic: Em projetos de produção, não use Base.metadata.create_all() para gerenciar alterações de colunas/tabelas; utilize o Alembic, que é a ferramenta oficial de migração do ecossistema SQLAlchemy.