Instalação

Para instalar o SQLAlchemy, execute o seguinte comando:

pip install sqlalchemy

Se precisar se conectar a um banco de dados específico, instale o driver correspondente:

# Para PostgreSQL
pip install psycopg2-binary

# Para MySQL
pip install mysql-connector-python

# Para SQLite, já está incluído na biblioteca padrão do Python.

Conceitos Fundamentais

Os componentes principais do SQLAlchemy ORM incluem:

• Motor (Engine): Responsável pela comunicação com o banco de dados.
• Sessão (Session): Gerencia as operações de persistência.
• Modelo (Model): Classes que mapeiam tabelas no banco de dados.
• Consulta (Query): Objeto para construir e executar consultas.

Conexão com o Banco de Dados

Crie uma instância de motor e uma fábrica de sessões:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Exemplo com SQLite
motor = create_engine('sqlite:///meu_banco.db', echo=False)

# Para PostgreSQL (descomente e ajuste conforme necessário)
# motor = create_engine('postgresql://usuario:senha@localhost:5432/meu_banco')

# Para MySQL
# motor = create_engine('mysql+mysqlconnector://usuario:senha@localhost:3306/meu_banco')

FabricaSessao = sessionmaker(autocommit=False, autoflush=False, bind=motor)
sessao = FabricaSessao()

Definição de Modelos de Dados

Utilize a base declarativa para definir os modelos:

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, index=True)
    nome_completo = Column(String(50), nullable=False)
    email_contato = Column(String(100), unique=True, index=True)
    
    # Relação um-para-muitos
    publicacoes = relationship("Publicacao", back_populates="autor")

class Publicacao(Base):
    __tablename__ = 'publicacoes'
    
    id_publicacao = Column(Integer, primary_key=True, index=True)
    titulo = Column(String(100), nullable=False)
    corpo = Column(String(500))
    autor_id = Column(Integer, ForeignKey('clientes.id_cliente'))
    
    # Relação muitos-para-um
    autor = relationship("Cliente", back_populates="publicacoes")
    
    # Relação muitos-para-muitos via tabela associativa
    etiquetas = relationship("Etiqueta", secondary="publicacao_etiquetas", back_populates="publicacoes")

class Etiqueta(Base):
    __tablename__ = 'etiquetas'
    
    id_etiqueta = Column(Integer, primary_key=True, index=True)
    rotulo = Column(String(30), unique=True, nullable=False)
    
    publicacoes = relationship("Publicacao", secondary="publicacao_etiquetas", back_populates="etiquetas")

class PublicacaoEtiqueta(Base):
    __tablename__ = 'publicacao_etiquetas'
    
    publicacao_id = Column(Integer, ForeignKey('publicacoes.id_publicacao'), primary_key=True)
    etiqueta_id = Column(Integer, ForeignKey('etiquetas.id_etiqueta'), primary_key=True)

Criação de Tabelas

Execute a criação das tabelas no banco de dados:

# Criar todas as tabelas
Base.metadata.create_all(bind=motor)

# Para remover todas as tabelas (use com cautela)
# Base.metadata.drop_all(bind=motor)

Operações CRUD Básicas

Criação de Dados

# Adicionar um novo cliente
novo_cliente = Cliente(nome_completo="Ana Silva", email_contato="ana@exemplo.com")
sessao.add(novo_cliente)
sessao.commit()

# Adicionar múltiplos registros
sessao.add_all([
    Cliente(nome_completo="Carlos Oliveira", email_contato="carlos@exemplo.com"),
    Cliente(nome_completo="Maria Souza", email_contato="maria@exemplo.com")
])
sessao.commit()

Leitura de Dados

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

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

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

Atualização de Dados

# Atualizar um registro existente
cliente = sessao.query(Cliente).get(1)
cliente.nome_completo = "Ana Silva Santos"
sessao.commit()

# Atualização em massa
sessao.query(Cliente).filter(Cliente.nome_completo.like("Ana%")).update({"nome_completo": "Ana S."}, synchronize_session=False)
sessao.commit()

Exclusão de Dados

# Excluir um registro
cliente = sessao.query(Cliente).get(1)
sessao.delete(cliente)
sessao.commit()

# Exclusão em massa
sessao.query(Cliente).filter(Cliente.nome_completo == "Carlos Oliveira").delete(synchronize_session=False)
sessao.commit()

Consultas Avançadas

Consultas Básicas

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

# Selecionar colunas específicas
nomes = sessao.query(Cliente.nome_completo).all()

# Ordenação
clientes_ordenados = sessao.query(Cliente).order_by(Cliente.nome_completo.desc()).all()

# Limitar resultados
clientes_limitados = sessao.query(Cliente).limit(5).all()

# Usar offset
clientes_com_offset = sessao.query(Cliente).offset(3).limit(5).all()

Filtros

from sqlalchemy import or_

# Filtro por igualdade
cliente = sessao.query(Cliente).filter(Cliente.nome_completo == "Ana Silva").first()

# Filtro com like
clientes = sessao.query(Cliente).filter(Cliente.nome_completo.like("%Silva%")).all()

# Filtro com IN
clientes = sessao.query(Cliente).filter(Cliente.nome_completo.in_(["Ana Silva", "Maria Souza"])).all()

# Múltiplas condições
clientes = sessao.query(Cliente).filter(
    Cliente.nome_completo == "Ana Silva",
    Cliente.email_contato.like("%@exemplo.com")
).all()

# Condição OR
clientes = sessao.query(Cliente).filter(
    or_(Cliente.nome_completo == "Ana Silva", Cliente.nome_completo == "Maria Souza")
).all()

# Diferente
clientes = sessao.query(Cliente).filter(Cliente.nome_completo != "Ana Silva").all()

Agregações

from sqlalchemy import func

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

# Contagem por grupo com junção
contagem_publicacoes = sessao.query(
    Cliente.nome_completo,
    func.count(Publicacao.id_publicacao)
).join(Publicacao).group_by(Cliente.nome_completo).all()

# Média
media_ids = sessao.query(func.avg(Cliente.id_cliente)).scalar()

Junções

# Junção interna
resultados = sessao.query(Cliente, Publicacao).join(Publicacao).filter(Publicacao.titulo.like("%Python%")).all()

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

# Condição de junção explícita
resultados = sessao.query(Cliente, Publicacao).join(Publicacao, Cliente.id_cliente == Publicacao.autor_id).all()

Operações com Relacionamentos

# Criar objetos relacionados
cliente = Cliente(nome_completo="Pedro Lima", email_contato="pedro@exemplo.com")
publicacao = Publicacao(titulo="Meu primeiro artigo", corpo="Conteúdo aqui", autor=cliente)
sessao.add(publicacao)
sessao.commit()

# Acessar relacionamentos
print(f"A publicação '{publicacao.titulo}' foi escrita por {publicacao.autor.nome_completo}")
print(f"Publicações do cliente {cliente.nome_completo}:")
for pub in cliente.publicacoes:
    print(f"  - {pub.titulo}")

# Manipular relacionamento muitos-para-muitos
etiqueta_python = Etiqueta(rotulo="Python")
etiqueta_sqlalchemy = Etiqueta(rotulo="SQLAlchemy")

publicacao.etiquetas.append(etiqueta_python)
publicacao.etiquetas.append(etiqueta_sqlalchemy)
sessao.commit()

print(f"Etiquetas da publicação '{publicacao.titulo}':")
for etq in publicacao.etiquetas:
    print(f"  - {etq.rotulo}")

Gerenciamento de Transações

# Transação com tratamento de erros
try:
    cliente = Cliente(nome_completo="Teste", email_contato="teste@exemplo.com")
    sessao.add(cliente)
    sessao.commit()
except Exception as e:
    sessao.rollback()
    print(f"Erro ocorrido: {e}")

# Função com gerenciamento de sessão
from sqlalchemy.orm import Session

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

# Transação aninhada
with sessao.begin_nested():
    cliente = Cliente(nome_completo="Transação Aninhada", email_contato="aninhado@exemplo.com")
    sessao.add(cliente)

# Savepoint
ponto_salvo = sessao.begin_nested()
try:
    cliente = Cliente(nome_completo="Savepoint", email_contato="savepoint@exemplo.com")
    sessao.add(cliente)
    ponto_salvo.commit()
except:
    ponto_salvo.rollback()

Boas Práticas

• Gerenciamento de Sessão: Crie uma nova sessão para cada unidade de trabalho e feche-a após o uso.
• Traatmento de Exceções: Sempre trate exceções e realize rollback quando necessário.
• Carregamento Preguiçoso: Evite problemas N+1 utilizando carrgeamento ansioso (eager loading) quando apropriado.
• Pool de Conexões: Configure adequadamente o tamenho e timeout do pool de conexões.
• Validação de Dados: Implemente validações no nível do modelo ou da aplicação para garantir integridade.

# Gerenciamento de sessão com context manager
from contextlib import contextmanager

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

# Exemplo de uso
with obter_sessao() as db:
    cliente = Cliente(nome_completo="Contexto", email_contato="contexto@exemplo.com")
    db.add(cliente)