Introdução

SQLAlchemy constitui um dos frameworks ORM (Object-Relational Mapping) mais utilizados no ecossistema Python, oferecendo mecanismos eficientes e flexíveis para interação com bancos de dados relacionais. Este material apresenta uma abordagem detalhada sobre as funcionalidades essenciais do SQLAlchemy ORM.

Sumário

1. Instalação do SQLAlchemy
2. Conceitos Fundamentais
3. Estabelecendo Conexão com Banco de Dados
4. Definindo Modelos de Dados
5. Criando Tabelas no Banco de Dados
6. Operações CRUD Básicas
7. Consultas ao Banco de Dados
8. Manipulação de Relacionamentos
9. Gerenciamento de Transações
10. Boas Práticas

Instalação

pip install sqlalchemy

Para conexão com bancos de dados específicos, é necessário instalar os drivers correspondentes:

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

# SQLite (Já incluído na biblioteca padrão Python)

Conceiots Fundamentais

• Engine: Motor responsável pela comunicação com o banco de dados
• Session: Sessão que gerencia todas as operações de persistência
• Model: Classe que representa uma tabela no banco de dados
• Query: Objeto utilizado para construir e executar consultas

Estabelecendo Conexão com Banco de Dados

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Criando o motor de conexão
# Exemplo com SQLite
engine = create_engine('sqlite:///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 a fábrica de sessões
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Instanciando uma sessão
sessao = SessionFactory()

Definindo Modelos de Dados

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

# Criando a classe base
Base = declarative_base()

class Cliente(Base):
    __tablename__ = 'clientes'
    
    id = Column(Integer, primary_key=True, index=True)
    nome = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)
    
    # Definindo relacionamento um-para-muitos
    pedidos = relationship("Pedido", back_populates="cliente")
    
class Pedido(Base):
    __tablename__ = 'pedidos'
    
    id = Column(Integer, primary_key=True, index=True)
    titulo = Column(String(100), nullable=False)
    descricao = Column(String(500))
    cliente_id = Column(Integer, ForeignKey('clientes.id'))
    
    # Definindo relacionamento muitos-para-um
    cliente = relationship("Cliente", back_populates="pedidos")
    
    # Definindo relacionamento muitos-para-muitos (via tabela associativa)
    categorias = relationship("Categoria", secondary="pedido_categorias", back_populates="pedidos")

class Categoria(Base):
    __tablename__ = 'categorias'
    
    id = Column(Integer, primary_key=True, index=True)
    nome = Column(String(30), unique=True, nullable=False)
    
    pedidos = relationship("Pedido", secondary="pedido_categorias", back_populates="categorias")

# Tabela associativa (para relacionamento muitos-para-muitos)
class PedidoCategoria(Base):
    __tablename__ = 'pedido_categorias'
    
    pedido_id = Column(Integer, ForeignKey('pedidos.id'), primary_key=True)
    categoria_id = Column(Integer, ForeignKey('categorias.id'), primary_key=True)

Criando Tabelas no Banco de Dados

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

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

Operações CRUD Básicas

Inserindo Dados

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

# Inserção em批量
sessao.add_all([
    Cliente(nome="Maria Santos", email="maria@exemplo.com"),
    Cliente(nome="Pedro Oliveira", email="pedro@exemplo.com")
])
sessao.commit()

Lendo Dados

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

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

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

Atualizando Dados

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

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

Excluindo Dados

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

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

Consultas ao Banco de Dados

Consultas Básicas

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

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

# Ordenação
clientes = sessao.query(Cliente).order_by(Cliente.nome.desc()).all()

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

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

Filtros em Consultas

from sqlalchemy import or_

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

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

# IN (busca em lista)
clientes = sessao.query(Cliente).filter(Cliente.nome.in_(["João Silva", "Maria Santos"])).all()

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

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

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

Consultas de Agregação

from sqlalchemy import func

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

# Agrupamento e contagem
contagem_pedidos = 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 JOIN

# Inner Join
resultados = sessao.query(Cliente, Pedido).join(Pedido).filter(Pedido.titulo.like("%Python%")).all()

# Left Outer Join
resultados = sessao.query(Cliente, Pedido).outerjoin(Pedido).all()

# Join com condição específica
resultados = sessao.query(Cliente, Pedido).join(Pedido, Cliente.id == Pedido.cliente_id).all()

Manipulação de Relacionamentos

# Criando objetos com relacionamentos
cliente = Cliente(nome="Ana Costa", email="ana@exemplo.com")
pedido = Pedido(titulo="Meu Primeiro Pedido", descricao="Olá Mundo!", cliente=cliente)
sessao.add(pedido)
sessao.commit()

# Acessandovia relacionamento
print(f"Pedido '{pedido.titulo}' pertence a {pedido.cliente.nome}")
print(f"Todos os pedidos do cliente {cliente.nome}:")
for p in cliente.pedidos:
    print(f"  - {p.titulo}")

# Operações em relacionamentos muitos-para-muitos
categoria_python = Categoria(nome="Python")
categoria_banco = Categoria(nome="Banco de Dados")

pedido.categorias.append(categoria_python)
pedido.categorias.append(categoria_banco)
sessao.commit()

print(f"Categorias do pedido '{pedido.titulo}':")
for cat in pedido.categorias:
    print(f"  - {cat.nome}")

Gerenciamento de Transações

# Commit automático com tratamento de exceções
try:
    cliente = Cliente(nome="Cliente Teste", email="teste@exemplo.com")
    sessao.add(cliente)
    sessao.commit()
except Exception as erro:
    sessao.rollback()
    print(f"Ocorreu um erro: {erro}")

# Usando gerenciador de contexto
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="Cliente Transação", email="transacao@exemplo.com")
    sessao.add(cliente)

# Savepoints (pontos de salvamento)
ponto_salvamento = sessao.begin_nested()
try:
    cliente = Cliente(nome="Cliente Savepoint", email="savepoint@exemplo.com")
    sessao.add(cliente)
    ponto_salvamento.commit()
except:
    ponto_salvamento.rollback()

Boas Práticas

1. Gerenciamento de Sessão: Criar uma nova sessão para cada requisição e fechá-la ao final
2. Tratamento de Exceções: Sempre tratar exceções e realizar rollback apropriado das transações
3. Carregamento Preguiçoso: Cuiddar com o problema N+1 de consultas, utilizar eager loading para otimização
4. Pool de Conexões: Configurar adequadamente o tamanho do pool e timeout de conexões
5. Validação de Dados: Validar integridade dos dados na camada de modelo ou aplicação

# Utilizando gerenciador de contexto para sessões
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="Cliente Contexto", email="contexto@exemplo.com")
    db.add(cliente)

Considerações Finais

SQLAlchemy ORM oferece um conjunto robusto e flexível de ferramentas para manipulação de banco de dados em Python. Após este material, você deverá ser capaz de:

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

SQLAlchemy possui diversas funcionalidades avançadas, como propreidades híbridas, listeners de eventos, e consultas personalizadas, que merecem ser exploradas em estudos posteriores.