Guia de Uso do SQLAlchemy ORM para Interação com Banco de Dados em Python

O SQLAlchemy é um dos frameworks ORM (Object-Relational Mapping) mais utilizados em Python, proporcionando uma maneira eficaz e flexível de interagir com bancos de dados. Este guia detalha como utilizar o SQLAlchemy ORM para diversas operações em banco de dados.

Instalação

Para instalar o SQLAlchemy, utilize o pip:


pip install sqlalchemy

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


# Para PostgreSQL
pip install psycopg2-binary

# Para MySQL
pip install mysql-connector-python

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

Conceitos Fundamentais

Engine: Gerencia a comunicação com o banco de dados.
Session: Mantém um conjunto de objetos mapeados e gerencia as operações de persistência.
Model: Representa uma tabela no banco de dados através de uma classse Python.
Query: Objeto utilizado para construir e executar consultas ao banco de dados.

Conectando ao Banco de Dados


from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Exemplo para SQLite: cria um arquivo de banco de dados chamado 'database.db'
engine = create_engine('sqlite:///database.db', echo=True)

# Exemplo para PostgreSQL
# engine = create_engine('postgresql://user:password@host:port/database_name')

# Exemplo para MySQL
# engine = create_engine('mysql+mysqlconnector://user:password@host:port/database_name')

# Cria uma fábrica de sessões
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Instancia uma sessão
session = SessionLocal()

Definnido Modelos de Dados

Os modelos herdam de uma classe base declarativa e definem as colunas da tabela.


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

# Cria a classe base para os modelos
Base = declarative_base()

class Aluno(Base):
    __tablename__ = 'alunos' # Nome da tabela no banco de dados
    
    id = Column(Integer, primary_key=True, index=True) # Chave primária
    nome = Column(String(100), nullable=False) # Coluna de nome, não pode ser nula
    email = Column(String(120), unique=True, index=True) # Email único e indexado
    
    # Relação um-para-muitos com Cursos
    cursos = relationship("Curso", back_populates="aluno") 

class Curso(Base):
    __tablename__ = 'cursos'
    
    id = Column(Integer, primary_key=True, index=True)
    titulo = Column(String(150), nullable=False)
    aluno_id = Column(Integer, ForeignKey('alunos.id')) # Chave estrangeira para Aluno
    
    # Relação muitos-para-um com Aluno
    aluno = relationship("Aluno", back_populates="cursos")

# Para relações muitos-para-muitos, é necessária uma tabela associativa
class Disciplina(Base):
    __tablename__ = 'disciplinas'
    
    id = Column(Integer, primary_key=True, index=True)
    nome = Column(String(50), unique=True, nullable=False)
    
    # Relação muitos-para-muitos com Alunos via tabela associativa
    alunos = relationship("Aluno", secondary="aluno_disciplinas", back_populates="disciplinas")

class AlunoDisciplina(Base):
    __tablename__ = 'aluno_disciplinas'
    
    aluno_id = Column(Integer, ForeignKey('alunos.id'), primary_key=True)
    disciplina_id = Column(Integer, ForeignKey('disciplinas.id'), primary_key=True)
    
    # Adicionando a relação de volta para o modelo Aluno para facilitar o acesso
    disciplina = relationship("Disciplina")

# Adicionando a relação 'disciplinas' ao modelo Aluno para relações M2M
Aluno.disciplinas = relationship("Disciplina", secondary="aluno_disciplinas", back_populates="alunos")

Criando Tabelas no Banco de Dados

Após definir os modelos, crie as tabelas correspondentes:


# Cria todas as tabelas definidas nos modelos
Base.metadata.create_all(bind=engine)

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

Operações CRUD Básicas

Inserindo Dados


# Cria um novo aluno
novo_aluno = Aluno(nome="Ana Silva", email="ana.silva@example.com")
session.add(novo_aluno)
session.commit() # Confirma a transação

# Inserindo múltiplos registros
session.add_all([
    Aluno(nome="Bruno Costa", email="bruno.costa@example.com"),
    Aluno(nome="Carla Dias", email="carla.dias@example.com")
])
session.commit()

Consultando Dados


# Busca todos os alunos
todos_alunos = session.query(Aluno).all()

# Busca o primeiro aluno encontrado
primeiro_aluno = session.query(Aluno).first()

# Busca um aluno específico pelo ID
aluno_por_id = session.query(Aluno).get(1)

Atualizando Dados


# Busca um aluno e atualiza seu nome
aluno_para_atualizar = session.query(Aluno).get(1)
if aluno_para_atualizar:
    aluno_para_atualizar.nome = "Ana Souza"
    session.commit()

# Atualização em lote (filtrando por nome e alterando para "Sr./Sra.")
session.query(Aluno).filter(Aluno.nome.like("Ana%")).update({"nome": "Sra. Ana"}, synchronize_session=False)
session.commit()

Excluindo Dados


# Busca um aluno e o remove
aluno_para_remover = session.query(Aluno).get(2)
if aluno_para_remover:
    session.delete(aluno_para_remover)
    session.commit()

# Exclusão em lote (remove alunos com email específico)
session.query(Aluno).filter(Aluno.email.endswith("@example.com")).delete(synchronize_session=False)
session.commit()

Consultas Avançadas

Filtragem


from sqlalchemy import or_

# Filtrar por nome exato
aluno_ana = session.query(Aluno).filter(Aluno.nome == "Ana Silva").first()

# Filtrar por padrão (emails contendo 'example.com')
emails_example = session.query(Aluno).filter(Aluno.email.like("%@example.com")).all()

# Filtrar por múltiplos valores (IN)
alunos_selecionados = session.query(Aluno).filter(Aluno.nome.in_(["Ana Silva", "Bruno Costa"])).all()

# Múltiplas condições (AND implícito)
aluno_especifico = session.query(Aluno).filter(Aluno.nome == "Ana Silva", Aluno.email != None).first()

# Condição OR
alunos_ou = session.query(Aluno).filter(or_(Aluno.nome == "Ana Silva", Aluno.nome == "Carla Dias")).all()

# Filtrar por não igual
alunos_nao_ana = session.query(Aluno).filter(Aluno.nome != "Ana Silva").all()

Ordenação, Limitação e Paginação


# Ordenar resultados por nome em ordem decrescente
alunos_ordenados = session.query(Aluno).order_by(Aluno.nome.desc()).all()

# Limitar o número de resultados
primeiros_cinco = session.query(Aluno).limit(5).all()

# Pular resultados (paginação)
pagina_2 = session.query(Aluno).offset(10).limit(10).all()

Agregações


from sqlalchemy import func

# Contar o número total de alunos
total_alunos = session.query(Aluno).count()

# Contar quantos cursos cada aluno possui
contagem_cursos_por_aluno = session.query(
    Aluno.nome, 
    func.count(Curso.id)
).join(Curso).group_by(Aluno.nome).all()

# Calcular a média de IDs dos alunos
media_ids = session.query(func.avg(Aluno.id)).scalar()

Joins (Junções)


# Inner Join: Buscar alunos e seus cursos
alunos_com_cursos = session.query(Aluno, Curso).join(Curso).filter(Curso.titulo.like("%Introdução%")).all()

# Left Outer Join: Buscar todos os alunos, e seus cursos se existirem
todos_alunos_e_cursos = session.query(Aluno, Curso).outerjoin(Curso).all()

# Join com condição explícita
alunos_e_seus_cursos = session.query(Aluno, Curso).join(Curso, Aluno.id == Curso.aluno_id).all()

# Join para relação Muitos-para-Muitos
alunos_e_disciplinas = session.query(Aluno, Disciplina).join(AlunoDisciplina).join(Disciplina).filter(Disciplina.nome == "Python").all()

Trabalhando com Relações


# Criar um aluno e associar um curso diretamente
aluno_novo = Aluno(nome="Pedro Alves", email="pedro.alves@example.com")
curso_novo = Curso(titulo="Ciência de Dados", aluno=aluno_novo) # Associa o curso ao aluno
session.add(curso_novo) # Adicionar o curso adiciona o aluno implicitamente devido à relação
session.commit()

# Acessar dados através das relações
print(f"O aluno {curso_novo.aluno.nome} está matriculado no curso '{curso_novo.titulo}'.")
print(f"Cursos de {aluno_novo.nome}:")
for c in aluno_novo.cursos:
    print(f"- {c.titulo}")

# Associando disciplinas (Muitos-para-Muitos)
disciplina_python = session.query(Disciplina).filter_by(nome="Python").first()
if not disciplina_python:
    disciplina_python = Disciplina(nome="Python")
    session.add(disciplina_python)
    session.commit()

# Adiciona a disciplina ao aluno
aluno_novo.disciplinas.append(disciplina_python) 
session.commit()

print(f"Disciplinas de {aluno_novo.nome}:")
for d in aluno_novo.disciplinas:
    print(f"- {d.nome}")

Gerenciamento de Transações

O SQLAlchemy gerencia transações automaticamente com session.commit() e session.rollback().


# Transação com tratamento de erro
try:
    aluno_transacao = Aluno(nome="Usuario Transacao", email="transacao@example.com")
    session.add(aluno_transacao)
    # Simula um erro para testar o rollback
    # raise Exception("Erro simulado") 
    session.commit() 
except Exception as e:
    session.rollback() # Desfaz as alterações se ocorrer um erro
    print(f"Erro na transação: {e}")

# Usando o gerenciador de contexto para transações
from sqlalchemy.orm import Session

def adicionar_curso(db: Session, titulo: str, aluno_id: int):
    try:
        novo_curso = Curso(titulo=titulo, aluno_id=aluno_id)
        db.add(novo_curso)
        db.commit()
        return novo_curso
    except:
        db.rollback()
        raise

# Transações aninhadas (savepoints)
with session.begin_nested() as nested_tx:
    aluno_savepoint = Aluno(nome="Aluno Savepoint", email="savepoint@example.com")
    session.add(aluno_savepoint)
    # Se ocorrer um erro aqui, apenas esta transação aninhada será revertida
    # nested_tx.rollback() # Descomente para testar o rollback aninhado

Boas Práticas

Gerenciamento de Sessão: Crie uma nova sessão para cada requisição e feche-a ao final.
Tratamento de Erros: Sempre use blocos try...except para commit e rollback.
Carregamento Otimizado (Eager Loading): Evite o problema N+1 usando joinedload ou selectinload quando necessário.
Pool de Conexões: Configure o tamanho do pool e tempos de expiração adequadamente.
Validação: Valide os dados antes de inseri-los ou atualizá-los no banco.

O SQLAlchemy oferece recursos avançados como eventos, mixins, consultas personalizadas e muito mais, proporcionando uma solução robusta parra ORM em Python.

Tags: SQLAlchemy Python ORM banco de dados CRUD

Publicado em 6-9 18:22 por Thomas

Doido Dev