SQLAlchemy representa uma das soluções ORM (Mapeamento Objeto-Relacional) mais robustas do ecossistema Python, oferecendo uma interface poderosa e flexível para interagir com bancos de dados. Este guia apresentará como utilizar o SQLAlchemy ORM para operações com bancos de dados.

Tabela de Conteúdo

1. Instalação do SQLAlchemy
2. Conceitos Fundamentais
3. Conexão com Banco de Dados
4. Definição de Modelos de Dados
5. Criação de Tabelas
6. Operações CRUD Básicas
7. Consultas de Dados
8. Operações com Relacionamentos
9. Geranciamento de Transações
10. Práticas Recomendadas

Instalação

Para começar, instale o SQLAlchemy através do pip:

bash

pip install sqlalchemy

Para conectar-se a bancos de dados específicos, instale os drivers correspondentes:

bash

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

# SQLite (já incluso na biblioteca padrão do Python)

Conceitos Fundamentais

• Engine: O mecanismo de conexão com o benco de dados, responsável pela comunicação
• Session: A sessão de banco de dados, gerenciando todas as operações de persistência
• Model: Classe de modelo de dados, representando tabelas no banco
• Query: Objeto de consulta, utilizado para construir e executar consultas

Conexão com Banco de Dados

python

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Criando o engine de conexão
# Exemplo com SQLite
engine = create_engine('sqlite:///banco_exemplo.db', echo=True)

# Exemplo com PostgreSQL
# engine = create_engine('postgresql://usuario:senha@localhost:5432/meubanco')

# Exemplo com MySQL
# engine = create_engine('mysql+mysqlconnector://usuario:senha@localhost:3306/meubanco')

# Criando fábrica de sessões
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Criando instância de sessão
sessao = SessionFactory()

Definição de Modelos de Dados

python

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

# Criando classe base
Base = declarative_base()

class Cliente(Base):
    __tablename__ = 'clientes'
    
    id = Column(Integer, primary_key=True, index=True)
    nome = Column(String(80), nullable=False)
    email = Column(String(120), unique=True, index=True)
    data_criacao = Column(DateTime, default=datetime.utcnow)
    
    # Relacionamento um para muitos
    pedidos = relationship("Pedido", back_populates="cliente")
    
class Produto(Base):
    __tablename__ = 'produtos'
    
    id = Column(Integer, primary_key=True, index=True)
    nome = Column(String(100), nullable=False)
    descricao = Column(String(300))
    preco = Column(Integer, nullable=False)
    estoque = Column(Integer, default=0)
    
    # Relacionamento muitos para muitos
    categorias = relationship("Categoria", secondary="produto_categorias", back_populates="produtos")

class Categoria(Base):
    __tablename__ = 'categorias'
    
    id = Column(Integer, primary_key=True, index=True)
    nome = Column(String(50), unique=True, nullable=False)
    
    produtos = relationship("Produto", secondary="produto_categorias", back_populates="categorias")

class Pedido(Base):
    __tablename__ = 'pedidos'
    
    id = Column(Integer, primary_key=True, index=True)
    data_pedido = Column(DateTime, default=datetime.utcnow)
    total = Column(Integer, nullable=False)
    cliente_id = Column(Integer, ForeignKey('clientes.id'))
    
    cliente = relationship("Cliente", back_populates="pedidos")
    itens = relationship("ItemPedido", back_populates="pedido")

class ItemPedido(Base):
    __tablename__ = 'itens_pedido'
    
    id = Column(Integer, primary_key=True, index=True)
    quantidade = Column(Integer, nullable=False)
    preco_unitario = Column(Integer, nullable=False)
    pedido_id = Column(Integer, ForeignKey('pedidos.id'))
    produto_id = Column(Integer, ForeignKey('produtos.id'))
    
    pedido = relationship("Pedido", back_populates="itens")
    produto = relationship("Produto")

# Tabela de associação (para relacionamento muitos para muitos)
class ProdutoCategoria(Base):
    __tablename__ = 'produto_categorias'
    
    produto_id = Column(Integer, ForeignKey('produtos.id'), primary_key=True)
    categoria_id = Column(Integer, ForeignKey('categorias.id'), primary_key=True)

Criação de Tabelas

python

# Criando todas as tabelas no banco
Base.metadata.create_all(bind=engine)

# Excluindo todas as tabelas
# Base.metadata.drop_all(bind=engine)

Operações CRUD Básicas

Criação de Dados

python

# Criando novo cliente
novo_cliente = Cliente(nome="João Silva", email="joao.silva@example.com")
sessao.add(novo_cliente)
sessao.commit()

# Criação em lote
sessao.add_all([
    Cliente(nome="Maria Oliveira", email="maria.oliveira@example.com"),
    Cliente(nome="Carlos Pereira", email="carlos.pereira@example.com")
])
sessao.commit()

Leitura de Dados

python

# Obtendo todos os clientes
clientes = sessao.query(Cliente).all()

# Obtendo o primeiro cliente
primeiro_cliente = sessao.query(Cliente).first()

# Buscando cliente por ID
cliente = sessao.query(Cliente).get(1)

Atualização de Dados

python

# Buscando e atualizando
cliente = sessao.query(Cliente).get(1)
cliente.nome = "João da Silva"
sessao.commit()

# Atualização em lote
sessao.query(Cliente).filter(Cliente.nome.like("João%")).update({"nome": "João Santos"}, synchronize_session=False)
sessao.commit()

Exclusão de Dados

python

# Buscando e excluindo
cliente = sessao.query(Cliente).get(1)
sessao.delete(cliente)
sessao.commit()

# Exclusão em lote
sessao.query(Cliente).filter(Cliente.nome == "Maria Oliveira").delete(synchronize_session=False)
sessao.commit()

Consultas de Dados

Consultas Simples

python

# Obtendo todos os registros
clientes = sessao.query(Cliente).all()

# Selecionando campos específicos
nomes = sessao.query(Cliente.nome).all()

# Ordenando resultados
clientes = sessao.query(Cliente).order_by(Cliente.nome.desc()).all()

# Limitando resultados
clientes = sessao.query(Cliente).limit(10).all()

# Paginação
clientes = sessao.query(Cliente).offset(5).limit(10).all()

Consultas com Filtros

python

from sqlalchemy import or_

# Filtro de igualdade
cliente = sessao.query(Cliente).filter(Cliente.nome == "João Silva").first()

# Busca com LIKE
clientes = sessao.query(Cliente).filter(Cliente.nome.like("João%")).all()

# Consulta IN
clientes = sessao.query(Cliente).filter(Cliente.nome.in_(["João Silva", "Maria Oliveira"])).all()

# Múltiplos critérios
clientes = sessao.query(Cliente).filter(
    Cliente.nome == "João Silva", 
    Cliente.email.like("%@example.com")
).all()

# Condição OR
clientes = sessao.query(Cliente).filter(
    or_(Cliente.nome == "João Silva", Cliente.nome == "Maria Oliveira")
).all()

# Diferente de
clientes = sessao.query(Cliente).filter(Cliente.nome != "João Silva").all()

Consultas Agregadas

python

from sqlalchemy import func

# Contagem
total_clientes = sessao.query(Cliente).count()

# Contagem agrupada
pedidos_por_cliente = sessao.query(
    Cliente.nome, 
    func.count(Pedido.id)
).join(Pedido).group_by(Cliente.nome).all()

# Média, soma, etc.
media_id = sessao.query(func.avg(Cliente.id)).scalar()

Consultas com Junções

python

# Junção interna
resultados = sessao.query(Cliente, Pedido).join(Pedido).filter(Pedido.total > 100).all()

# Junção externa esquerda
resultados = sessao.query(Cliente, Pedido).outerjoin(Pedido).all()

# Especificando condição de junção
resultados = sessao.query(Cliente, Pedido).join(Pedido, Cliente.id == Pedido.cliente_id).all()

Operações com Relacionamentos

python

# Criando objetos com relacionamentos
cliente = Cliente(nome="Ana Costa", email="ana.costa@example.com")
produto = Produto(nome="Notebook", descricao="Laptop高性能", preco=5000, estoque=10)
pedido = Pedido(total=5000, cliente=cliente, itens=[ItemPedido(quantidade=1, preco_unitario=5000, produto=produto)])

sessao.add(pedido)
sessao.commit()

# Acessando relacionamentos
print(f"Pedido #{pedido.id} pertence a {pedido.cliente.nome}")
print(f"Cliente {cliente.nome} tem {len(cliente.pedidos)} pedidos:")

for p in cliente.pedidos:
    print(f"  - Pedido #{p.id} totalizando R${p.total}")

# Relacionamento muitos para muitos
cat_eletronicos = Categoria(nome="Eletrônicos")
cat_computadores = Categoria(nome="Computadores")

produto.categorias.append(cat_eletronicos)
produto.categorias.append(cat_computadores)
sessao.commit()

print(f"Produto '{produto.nome}' pertence às categorias:")
for categoria in produto.categorias:
    print(f"  - {categoria.nome}")

Gerenciamento de Transações

python

# Transação com commit automático
try:
    cliente = Cliente(nome="Usuário Teste", email="teste@example.com")
    sessao.add(cliente)
    sessao.commit()
except Exception as e:
    sessao.rollback()
    print(f"Ocorreu um erro: {e}")

# Usando gerenciador de contexto de transação
from sqlalchemy.orm import Session

def criar_cliente(sessao: Session, nome: str, email: str):
    try:
        cliente = Cliente(nome=nome, email=email)
        sessao.add(cliente)
        sessao.commit()
        return cliente
    except:
        sessao.rollback()
        raise

# Transações aninhadas
with sessao.begin_nested():
    cliente = Cliente(nome="Usuário Transação", email="transacao@example.com")
    sessao.add(cliente)

# Pontos de salvamento
ponto_salvamento = sessao.begin_nested()
try:
    cliente = Cliente(nome="Usuário Salvamento", email="salvamento@example.com")
    sessao.add(cliente)
    ponto_salvamento.commit()
except:
    ponto_salvamento.rollback()

Práticas Recomendadas

1. Gerenciamento de Sessão: Crie uma nova sessão para cada requisição e feche ao final
2. Tratamento de Exceções: Sempre trate exceções e faça rollback apropriado das transações
3. Carregamento Precoce: Cuidado com o problema de N+1 consultas, use eager loading para otimizar
4. Pool de Conexões: Configure adequadamente o tamanho do pool de conexões e timeouts
5. Validação de Dados: Implemente validação de integridade de dados na camada de modelo ou aplicação

python

# Gerenciamento de sessão com gerenciador de contexto
from contextlib import contextmanager

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

# Exemplo de uso
with obter_sessao() as db:
    cliente = Cliente(nome="Usuário Contexto", email="contexto@example.com")
    db.add(cliente)

Conclusão

O SQLAlchemy ORM oferece uma abordagem poderosa e flexível para operações com bancos de dados. Com este guia, você deve ser capaz de:

1. Instalar e configurar o SQLAlchemy
2. Definir modelos de dados e relacionamentos
3. Executar operações CRUD básicas
4. Construir consultas complexas
5. Gerenciar transações de banco de dados
6. Seguir as práticas recomendadas

O SQLAlchemy possui muitas características avançadas, como propriedades híbridas, escutas de eventos e consultas personalizadas, que valem a pena explorar para aprofundar seu conhecimento.