Instalação e Confgiuração

Para utilziar o SQLAlchemy, primeiro instale-o via pip:

pip install sqlalchemy

Instale também o driver do banco de dados específico, caso necessário:

# Para PostgreSQL
pip install psycopg2-binary

# Para MySQL
pip install mysql-connector-python

Estabelecendo a Conexão com o Banco

Crie uma instância do mecanismo de conexão e uma fábrica de sessões:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Configurar o mecanismo de conexão
meu_engine = create_engine('sqlite:///meubanco.db', echo=False)

# Definir a fábrica de sessão
SessaoLocal = sessionmaker(bind=meu_engine)

# Obter uma sessão
sessao_atual = SessaoLocal()

Modelagem dos Dados

Defina as classes que representam suas tabelas usando o sistema declarativo:

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

Base = declarative_base()

class Cliente(Base):
    __tablename__ = 'clientes'
    
    id_cliente = Column(Integer, primary_key=True)
    nome_completo = Column(String(100), nullable=False)
    endereco_email = Column(String(150), unique=True)
    
    pedidos = relationship("Pedido", back_populates="cliente")

class Produto(Base):
    __tablename__ = 'produtos'
    
    codigo_produto = Column(Integer, primary_key=True)
    descricao = Column(String(200), nullable=False)
    preco_unitario = Column(Integer)

class Pedido(Base):
    __tablename__ = 'pedidos'
    
    numero_pedido = Column(Integer, primary_key=True)
    id_cliente = Column(Integer, ForeignKey('clientes.id_cliente'))
    data_criacao = Column(String(20))
    
    cliente = relationship("Cliente", back_populates="pedidos")
    itens_pedido = relationship("ItemPedido", back_populates="pedido")

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

Criação das Tabelas no Banco

Base.metadata.create_all(bind=meu_engine)

Operações de Inserção

novo_cliente = Cliente(nome_completo="Ana Silva", endereco_email="ana@email.com")
sessao_atual.add(novo_cliente)
sessao_atual.commit()

# Inserção em lote
lista_clientes = [
    Cliente(nome_completo="Carlos Souza", endereco_email="carlos@email.com"),
    Cliente(nome_completo="Maria Oliveira", endereco_email="maria@email.com")
]
sessao_atual.add_all(lista_clientes)
sessao_atual.commit()

Consultas ao Banco de Dados

# Buscar todos os registros
todos_clientes = sessao_atual.query(Cliente).all()

# Filtrar por condições
cliente_especifico = sessao_atual.query(Cliente).filter(
    Cliente.nome_completo.like("%Silva%")
).first()

# Ordenação e limitação
clientes_recentes = sessao_atual.query(Cliente).order_by(
    Cliente.id_cliente.desc()
).limit(10).all()

Modificação e Remoção de Dados

# Atualizar um registro
registro = sessao_atual.query(Cliente).get(1)
registro.endereco_email = "novo_email@email.com"
sessao_atual.commit()

# Exclusão de registro
sessao_atual.delete(registro)
sessao_atual.commit()

Tratamento de Transações

try:
    novo_pedido = Pedido(id_cliente=5, data_criacao="2023-10-01")
    sessao_atual.add(novo_pedido)
    sessao_atual.flush()
    
    item = ItemPedido(pedido_id=novo_pedido.numero_pedido, produto_id=10, quantidade=2)
    sessao_atual.add(item)
    
    sessao_atual.commit()
except Exception as erro:
    sessao_atual.rollback()
    print(f"Erro na transação: {erro}")

Gerenciamento de Sessões com Contexto

from contextlib import contextmanager

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

# Utilização
with obter_sessao() as db:
    novo_registro = Cliente(nome_completo="Teste", endereco_email="teste@email.com")
    db.add(novo_registro)

Dicas de Desempenho

Para otimizar consultas, utilize eager loading quando for acessar relacionamentos:

from sqlalchemy.orm import joinedload

resultados = sessao_atual.query(Cliente).options(
    joinedload(Cliente.pedidos)
).all()