O framework SQLAlchemy consolidou-se como a solução padrão da indústria para mapeamento objeto-relacional (ORM) e gerenciamento de bancos de dados em Python. A versão 2.0 entroduziu uma abordagem orientada a tipagem estática e consultas baseadas em expressões, oferecendo maior perofrmance e segurança de tipos.

Configuração do Ambiente

A instalação do núcleo do framework é realizada via pip. Dependendo do SGBD alvo, drivers específicos devem ser incluídos.

# Instalação do núcleo
pip install sqlalchemy

# Drivers opcionais para SGBDs específicos
pip install psycopg2-binary  # PostgreSQL
pip install pymysql          # MySQL

Arquitetura e Conceitos Fundamentais

• Engine: Ponto de partida para a comunicação com o banco, gerenciando o pool de conexões e a emissão de instruções SQL.
• Session: Workspace que rastreia alterações em objetos e coordena transações com o banco de dados.
• Mapper: Componente responsável por traduzir classes Python para tabelas relacionais e vice-versa.

Estabelecendo a Conexão

A engine é configurada utilizando uma string de conexão (URL). O parâmetro echo pode ser ativado para depuração de SQL.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Configuração para SQLite em memória (ideal para testes)
engine = create_engine("sqlite+pysqlite:///:memory:", echo=False, future=True)

# Configuração para PostgreSQL (exemplo)
# engine = create_engine("postgresql+psycopg2://user:pass@localhost/dbname", pool_size=10)

SessionFactory = sessionmaker(bind=engine, expire_on_commit=False)

Definição de Modelos com Tipagem Estática (Estilo 2.0)

A nova API utiliza anotações de tipo do Python para definir colunas, eliminando a necessidade do construtor Column explícito e melhorando a integração com mypy.

from sqlalchemy import String, Integer, ForeignKey, Table
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class Cliente(Base):
    __tablename__ = "clientes"

    id: Mapped[int] = mapped_column(primary_key=True)
    nome: Mapped[str] = mapped_column(String(120), nullable=False)
    email: Mapped[str] = mapped_column(String(150), unique=True, index=True)
    
    pedidos: Mapped[list["Pedido"]] = relationship(back_populates="comprador", cascade="all, delete-orphan")

class Pedido(Base):
    __tablename__ = "pedidos"

    id: Mapped[int] = mapped_column(primary_key=True)
    descricao: Mapped[str] = mapped_column(String(255))
    valor_total: Mapped[float] = mapped_column(default=0.0)
    cliente_id: Mapped[int] = mapped_column(ForeignKey("clientes.id"))
    
    comprador: Mapped["Cliente"] = relationship(back_populates="pedidos")
    etiquetas: Mapped[list["Etiqueta"]] = relationship(secondary="pedido_etiquetas", back_populates="pedidos")

class Etiqueta(Base):
    __tablename__ = "etiquetas"

    id: Mapped[int] = mapped_column(primary_key=True)
    nome: Mapped[str] = mapped_column(String(50), unique=True)
    
    pedidos: Mapped[list["Pedido"]] = relationship(secondary="pedido_etiquetas", back_populates="etiquetas")

# Tabela de associação para relacionamento N:N
pedido_etiquetas = Table(
    "pedido_etiquetas",
    Base.metadata,
    mapped_column("pedido_id", ForeignKey("pedidos.id"), primary_key=True),
    mapped_column("etiqueta_id", ForeignKey("etiquetas.id"), primary_key=True)
)

Sincronização do Esquema

A criação das tabelas no banco de dados é executada através do metadados da classe base.

Base.metadata.create_all(bind=engine)

Operações CRUD com a API 2.0

Inserção de Registros

from sqlalchemy import select

with SessionFactory() as session:
    novo_cliente = Cliente(nome="Ana Souza", email="ana.souza@dominio.com")
    session.add(novo_cliente)
    session.commit()
    
    # Inserção em lote
    clientes_lote = [
        Cliente(nome="Carlos Lima", email="carlos@dominio.com"),
        Cliente(nome="Beatriz Costa", email="bia@dominio.com")
    ]
    session.add_all(clientes_lote)
    session.commit()

Lietura e Consultas

A API 2.0 substitui o legado session.query() por session.execute(select()).

with SessionFactory() as session:
    # Recuperar todos os registros
    stmt = select(Cliente)
    clientes = session.scalars(stmt).all()

    # Recuperar por chave primária
    stmt = select(Cliente).where(Cliente.id == 1)
    cliente_alvo = session.scalar(stmt)

    # Projeção de colunas específicas
    stmt = select(Cliente.nome, Cliente.email).where(Cliente.nome.like("A%"))
    resultados = session.execute(stmt).all()

Atualização e Remoção

from sqlalchemy import update, delete

with SessionFactory() as session:
    # Atualização via objeto
    stmt = select(Cliente).where(Cliente.id == 1)
    cliente = session.scalar(stmt)
    if cliente:
        cliente.nome = "Ana Souza Lima"
        session.commit()

    # Atualização em lote (Bulk)
    stmt = update(Cliente).where(Cliente.nome.like("C%")).values(nome="Cliente Atualizado")
    session.execute(stmt)
    session.commit()

    # Remoção em lote
    stmt = delete(Cliente).where(Cliente.email == "bia@dominio.com")
    session.execute(stmt)
    session.commit()

Consultas Avançadas e Agregações

from sqlalchemy import func, or_

with SessionFactory() as session:
    # Filtros complexos e OR
    stmt = select(Cliente).where(
        or_(Cliente.nome == "Ana Souza Lima", Cliente.email.like("%@dominio.com"))
    )
    
    # Joins e Agregações
    stmt = (
        select(Cliente.nome, func.count(Pedido.id).label("total_pedidos"))
        .join(Pedido, Cliente.id == Pedido.cliente_id)
        .group_by(Cliente.nome)
        .having(func.count(Pedido.id) > 0)
    )
    resumo = session.execute(stmt).all()
    
    # Ordenação e Paginação
    stmt = select(Cliente).order_by(Cliente.nome.desc()).offset(10).limit(5)
    pagina = session.scalars(stmt).all()

Gerenciamento de Transações e Savepoints

O controle transacional garante a integridade dos dados, permitindo aninhamento através de savepoints.

with SessionFactory() as session:
    try:
        with session.begin():
            cliente = Cliente(nome="Teste Transacional", email="teste@dominio.com")
            session.add(cliente)
            
            # Savepoint aninhado
            with session.begin_nested():
                pedido = Pedido(descricao="Pedido de Teste", valor_total=150.0, comprador=cliente)
                session.add(pedido)
                # Se falhar aqui, apenas o pedido é revertido, o cliente é mantido
                
    except Exception as e:
        print(f"Falha na transação: {e}")
        # O rollback é automático ao sair do bloco 'with session.begin()' em caso de exceção

Padrões de Projeto e Ciclo de Vida da Sessão

Para aplicações web ou APIs, a sessão deve ter um escopo bem definido, geralmente vinculado ao ciclo de vida de uma requisição.

from contextlib import contextmanager

@contextmanager
def get_transactional_session():
    session = SessionFactory()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Utilização em camadas de serviço
def processar_venda(dados_cliente, dados_pedido):
    with get_transactional_session() as db:
        cliente = Cliente(**dados_cliente)
        db.add(cliente)
        db.flush() # Garante o ID do cliente sem fechar a transação
        
        pedido = Pedido(**dados_pedido, cliente_id=cliente.id)
        db.add(pedido)

Para mitigar o problema de consultas N+1, utilize a estratégia de carregamento antecipado (eager loading) através de selectinload ou joinedload nas relações.