SQLAlchemy ORM: mapeamento de objetos

1. Introdução ao SQLAlchemy ORM

O SQLAlchemy é um dos ORMs (Object-Relational Mapping) mais poderosos e flexíveis do ecossistema Python. Um ORM permite que você interaja com bancos de dados relacionais usando objetos Python, abstraindo a complexidade do SQL puro. O SQLAlchemy oferece duas abordagens principais: o SQLAlchemy Core, que fornece uma camada de abstração sobre SQL, e o SQLAlchemy ORM, que mapeia classes Python para tabelas do banco de dados.

A principal vantagem do ORM é que você pode trabalhar com objetos familiares ao Python, enquanto o SQLAlchemy gerencia a tradução para comandos SQL. Isso aumenta a produtividade e reduz a probabilidade de erros de sintaxe SQL.

Para instalar o SQLAlchemy, utilize:

pip install sqlalchemy

Para este artigo, usaremos SQLite como banco de dados, mas o SQLAlchemy suporta PostgreSQL, MySQL, Oracle e outros.

2. Configuração da Engine e Sessão

A engine é o ponto de partida para qualquer aplicação SQLAlchemy. Ela gerencia conexões com o banco de dados.

from sqlalchemy import create_engine

# Criando engine para SQLite (arquivo local)
engine = create_engine('sqlite:///meu_banco.db', echo=True)

O parâmetro echo=True exibe as consultas SQL geradas no console, útil para depuração.

Para gerenciar sessões, usamos sessionmaker():

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)
session = Session()

A melhor prática é usar context managers para garantir que a sessão seja fechada adequadamente:

from contextlib import contextmanager

@contextmanager
def get_session():
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()

3. Declarando Modelos com Mapeamento

O SQLAlchemy ORM usa declarative_base() para criar classes que representam tabelas:

from sqlalchemy import Column, Integer, String, Float, ForeignKey
from sqlalchemy.orm import declarative_base, relationship

Base = declarative_base()

class Usuario(Base):
    __tablename__ = 'usuarios'

    id = Column(Integer, primary_key=True)
    nome = Column(String(100), nullable=False)
    email = Column(String(150), unique=True, index=True)
    idade = Column(Integer)

    # Relacionamento
    pedidos = relationship('Pedido', back_populates='usuario')

    def __repr__(self):
        return f'<Usuario(nome={self.nome})>'

class Pedido(Base):
    __tablename__ = 'pedidos'

    id = Column(Integer, primary_key=True)
    descricao = Column(String(200))
    valor = Column(Float)
    usuario_id = Column(Integer, ForeignKey('usuarios.id'))

    usuario = relationship('Usuario', back_populates='pedidos')

    def __repr__(self):
        return f'<Pedido(descricao={self.descricao})>'

Criamos as tabelas no banco com:

Base.metadata.create_all(engine)

4. Operações CRUD com a Sessão

Inserindo dados

with get_session() as session:
    # Adicionando um único objeto
    usuario = Usuario(nome='João Silva', email='joao@email.com', idade=30)
    session.add(usuario)

    # Adicionando múltiplos objetos
    usuarios = [
        Usuario(nome='Maria Santos', email='maria@email.com', idade=25),
        Usuario(nome='Pedro Costa', email='pedro@email.com', idade=35)
    ]
    session.add_all(usuarios)

Consultando dados

with get_session() as session:
    # Todos os usuários
    todos = session.query(Usuario).all()

    # Filtros
    usuarios_jovens = session.query(Usuario).filter(Usuario.idade < 30).all()

    # Ordenação e limites
    usuarios_ordenados = session.query(Usuario).order_by(Usuario.nome).limit(5).all()

    # Primeiro resultado
    usuario = session.query(Usuario).filter_by(email='joao@email.com').first()

Atualizando e excluindo

with get_session() as session:
    # Atualização
    usuario = session.query(Usuario).filter_by(id=1).first()
    usuario.idade = 31

    # Exclusão
    session.delete(usuario)

5. Relacionamentos Avançados entre Tabelas

Um-para-muitos (já demonstrado acima)

Um-para-um

class Perfil(Base):
    __tablename__ = 'perfis'

    id = Column(Integer, primary_key=True)
    bio = Column(String(500))
    usuario_id = Column(Integer, ForeignKey('usuarios.id'), unique=True)

    usuario = relationship('Usuario', back_populates='perfil')

# Adicionando ao modelo Usuario
# perfil = relationship('Perfil', back_populates='usuario', uselist=False)

Muitos-para-muitos

from sqlalchemy import Table, Column

# Tabela de associação
produto_pedido = Table(
    'produto_pedido', Base.metadata,
    Column('produto_id', Integer, ForeignKey('produtos.id')),
    Column('pedido_id', Integer, ForeignKey('pedidos.id'))
)

class Produto(Base):
    __tablename__ = 'produtos'

    id = Column(Integer, primary_key=True)
    nome = Column(String(100))

    pedidos = relationship('Pedido', secondary=produto_pedido, back_populates='produtos')

class Pedido(Base):
    # ... (código anterior)
    produtos = relationship('Produto', secondary=produto_pedido, back_populates='pedidos')

Carregamento lazy vs eager

# Lazy (padrão) - carrega quando acessado
pedido = session.query(Pedido).first()
print(pedido.produtos)  # SQL executado aqui

# Eager com joined
from sqlalchemy.orm import joinedload
pedido = session.query(Pedido).options(joinedload(Pedido.produtos)).first()

6. Consultas Avançadas e Junções

Filtros compostos

from sqlalchemy import and_, or_

with get_session() as session:
    # AND
    usuarios = session.query(Usuario).filter(
        and_(Usuario.idade >= 20, Usuario.idade <= 40)
    ).all()

    # OR
    usuarios = session.query(Usuario).filter(
        or_(Usuario.nome.like('João%'), Usuario.email.like('%@email.com'))
    ).all()

    # IN
    usuarios = session.query(Usuario).filter(Usuario.id.in_([1, 2, 3])).all()

Junções

with get_session() as session:
    # Junção explícita
    resultados = session.query(Usuario, Pedido).join(Pedido).all()

    # Outer join
    resultados = session.query(Usuario, Pedido).outerjoin(Pedido).all()

    # Junção com filtro
    usuarios_com_pedidos = session.query(Usuario).join(Pedido).filter(
        Pedido.valor > 100
    ).all()

Agregações

from sqlalchemy import func

with get_session() as session:
    # Contagem
    total_usuarios = session.query(func.count(Usuario.id)).scalar()

    # Soma e agrupamento
    resultados = session.query(
        Usuario.nome,
        func.sum(Pedido.valor).label('total_gasto')
    ).join(Pedido).group_by(Usuario.id).all()

7. Transações, Flush e Controle de Estado

Controle de transações

session = Session()
try:
    usuario1 = Usuario(nome='Ana', email='ana@email.com')
    session.add(usuario1)

    usuario2 = Usuario(nome='Carlos', email='carlos@email.com')
    session.add(usuario2)

    # Confirma a transação
    session.commit()
except:
    # Reverte em caso de erro
    session.rollback()
    raise
finally:
    session.close()

Diferença entre flush() e commit()

session = Session()
usuario = Usuario(nome='Teste', email='teste@email.com')
session.add(usuario)

# flush() envia para o banco mas não confirma
session.flush()
print(usuario.id)  # Agora tem ID, mesmo sem commit

# commit() confirma permanentemente
session.commit()

Estados de objeto

# Transient - objeto criado mas não associado à sessão
novo_usuario = Usuario(nome='Novo', email='novo@email.com')

# Pending - adicionado à sessão mas não persistido
session.add(novo_usuario)

# Persistent - persistido no banco
session.flush()

# Detached - sessão foi fechada
session.close()

8. Boas Práticas e Considerações Finais

Migrações com Alembic

Para ambientes de produção, use Alembic para gerenciar migrações:

# Instalação
pip install alembic

# Inicialização
alembic init alembic

# Criar migração
alembic revision --autogenerate -m "criar_tabelas"

# Aplicar migração
alembic upgrade head

Performance

Para otimizar consultas com relacionamentos:

from sqlalchemy.orm import selectinload, subqueryload

# selectinload() - carrega em consulta separada com IN
usuarios = session.query(Usuario).options(
    selectinload(Usuario.pedidos)
).all()

# subqueryload() - carrega usando subquery
usuarios = session.query(Usuario).options(
    subqueryload(Usuario.pedidos)
).all()

Tratamento de erros

from sqlalchemy.exc import IntegrityError, SQLAlchemyError

try:
    with get_session() as session:
        usuario = Usuario(nome='João', email='joao@email.com')
        session.add(usuario)
except IntegrityError:
    print("Erro: email já existe")
except SQLAlchemyError as e:
    print(f"Erro no banco de dados: {e}")

O SQLAlchemy ORM oferece um equilíbrio perfeito entre produtividade e controle. Comece com modelos simples e vá adicionando complexidade conforme necessário. Lembre-se de sempre fechar as sessões e usar transações para garantir a integridade dos dados.

Referências