O SQLAlchemy é um dos frameworks mais populares de ORM (Mapeamento Objeto-Relacional) em Python, oferecendo uma forma eficiente e flexível de operações com banco de dados. Este guia demonstrará como utilizar o SQLAlchemy ORM para manipulação de dados.

Conteúdo

1. Instalação do SQLAlchemy
2. Conceitos fundamentais
3. Conexão com o banco de dados
4. Definição de modelos de dados
5. Criação de tabelas no banco
6. Operações CRUD básicas
7. Consulta de dados
8. Operações com relacionamentos
9. Gerenciamento de transações
10. Práticas recomendadas

Instalação

bash

pip install sqlalchemy

Para conexão com bancos 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 banco de dados, responsável pela comunicação
• Session: Sessão do banco, gerencia todas as operações de persistência
• Model: Clase de modelo de dados, correspondente a uma tabela no banco
• Query: Objeto de consulta, usado para construir e executar consultas

Conexão com o banco de dados

python

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

 # Criando o motor 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 a 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
from sqlalchemy.orm import relationship, declarative_base

 # Criando classe base
Base = declarative_base()

class Funcionario(Base):
    __tablename__ = 'funcionarios'
    
    id = Column(Integer, primary_key=True, index=True)
    nome = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)
    
    # Definindo relação um-para-muitos
    projetos = relationship("Projeto", back_populates="lider")

class Projeto(Base):
    __tablename__ = 'projetos'
    
    id = Column(Integer, primary_key=True, index=True)
    nome = Column(String(100), nullable=False)
    descricao = Column(String(500))
    lider_id = Column(Integer, ForeignKey('funcionarios.id'))
    
    # Definindo relação muitos-para-um
    lider = relationship("Funcionario", back_populates="projetos")
    
    # Definindo relação muitos-para-muitos (através de tabela de junção)
    habilidades = relationship("Habilidade", secondary="projeto_habilidades", back_populates="projetos")

class Habilidade(Base):
    __tablename__ = 'habilidades'
    
    id = Column(Integer, primary_key=True, index=True)
    nome = Column(String(30), unique=True, nullable=False)
    
    projetos = relationship("Projeto", secondary="projeto_habilidades", back_populates="habilidades")

 # Tabela de junção (para relações muitos-para-muitos)
class ProjetoHabilidade(Base):
    __tablename__ = 'projeto_habilidades'
    
    projeto_id = Column(Integer, ForeignKey('projetos.id'), primary_key=True)
    habilidade_id = Column(Integer, ForeignKey('habilidades.id'), primary_key=True)

Criação de tabelas no banco

python

 # Criando todas as tabelas
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 funcionário
novo_funcionario = Funcionario(nome="João Silva", email="joao.silva@empresa.com")
sessao.add(novo_funcionario)
sessao.commit()

 # Criação em lote
sessao.add_all([
    Funcionario(nome="Maria Santos", email="maria.santos@empresa.com"),
    Funcionario(nome="Pedro Oliveira", email="pedro.oliveira@empresa.com")
])
sessao.commit()

Leitura de dados

python

 # Obtendo todos os funcionários
funcionarios = sessao.query(Funcionario).all()

 # Obtendo o primeiro funcionário
primeiro_funcionario = sessao.query(Funcionario).first()

 # Obtendo funcionário por ID
funcionario = sessao.query(Funcionario).get(1)

Atualização de dados

python

 # Consulta e atualização
funcionario = sessao.query(Funcionario).get(1)
funcionario.nome = "João Silva Santos"
sessao.commit()

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

Exclusão de dados

python

 # Consulta e exclusão
funcionario = sessao.query(Funcionario).get(1)
sessao.delete(funcionario)
sessao.commit()

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

Consulta de dados

Consultas básicas

python

 # Obter todos os registros
funcionarios = sessao.query(Funcionario).all()

 # Obter campos específicos
nomes = sessao.query(Funcionario.nome).all()

 # Ordenação
funcionarios = sessao.query(Funcionario).order_by(Funcionario.nome.desc()).all()

 # Limitar quantidade de resultados
funcionarios = sessao.query(Funcionario).limit(10).all()

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

Filtragem de consultas

python

from sqlalchemy import or_

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

 # Busca por padrão
funcionarios = sessao.query(Funcionario).filter(Funcionario.nome.like("João%")).all()

 # Busca IN
funcionarios = sessao.query(Funcionario).filter(Funcionario.nome.in_(["João Silva", "Maria Santos"])).all()

 # Múltiplos filtros
funcionarios = sessao.query(Funcionario).filter(
    Funcionario.nome == "João Silva", 
    Funcionario.email.like("%@empresa.com")
).all()

 # Condição OU
funcionarios = sessao.query(Funcionario).filter(
    or_(Funcionario.nome == "João Silva", Funcionario.nome == "Maria Santos")
).all()

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

Consultas agregadas

python

from sqlalchemy import func

 # Contagem
total = sessao.query(Funcionario).count()

 # Contagem agrupada
contagem_projetos = sessao.query(
    Funcionario.nome, 
    func.count(Projeto.id)
).join(Projeto).group_by(Funcionario.nome).all()

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

Consultas com junções

python

 # Junção interna
resultados = sessao.query(Funcionario, Projeto).join(Projeto).filter(Projeto.nome.like("%Python%")).all()

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

 # Especificando condição de junção
resultados = sessao.query(Funcionario, Projeto).join(Projeto, Funcionario.id == Projeto.lider_id).all()

Operações com relacionamentos

python

 # Criando objetos com relacionamentos
funcionario = Funcionario(nome="Carlos Mendes", email="carlos.mendes@empresa.com")
projeto = Projeto(nome="Sistema de Gestão", descricao="Sistema interno para gestão de processos", lider=funcionario)
sessao.add(projeto)
sessao.commit()

 # Acessando através de relacionamentos
print(f"O projeto '{projeto.nome}' é liderado por {projeto.lider.nome}")
print(f"O funcionário {funcionario.nome} participa dos seguintes projetos:")
for p in funcionario.projetos:
    print(f"  - {p.nome}")

 # Operações com relacionamentos muitos-para-muitos
python_habilidade = Habilidade(nome="Python")
django_habilidade = Habilidade(nome="Django")

projeto.habilidades.append(python_habilidade)
projeto.habilidades.append(django_habilidade)
sessao.commit()

print(f"O projeto '{projeto.nome}' requer as seguintes habilidades:")
for habilidade in projeto.habilidades:
    print(f"  - {habilidade.nome}")

Gerenciamento de transações

python

 # Transação com commit automático
try:
    funcionario = Funcionario(nome="Usuário Teste", email="teste@empresa.com")
    sessao.add(funcionario)
    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_funcionario(sessao: Session, nome: str, email: str):
    try:
        funcionario = Funcionario(nome=nome, email=email)
        sessao.add(funcionario)
        sessao.commit()
        return funcionario
    except:
        sessao.rollback()
        raise

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

 # Ponto de salvamento
ponto_salvamento = sessao.begin_nested()
try:
    funcionario = Funcionario(nome="Usuário Ponto", email="ponto@empresa.com")
    sessao.add(funcionario)
    ponto_salvamento.commit()
except:
    ponto_salvamento.rollback()

Práticas recomendadas

1. Gerenciamento de sessão: Criar nova sessão para cada requisição e fechá-la ao final
2. Tratamento de exceções: Sempre tratar exceções e fazer rollback de transações apropriado
3. Carregamento preguiçoso: Cuidar do problema de N+1 consultas, usando carregamento ansioso
4. Pool de conexões: Configurar adequadamente tamanho e timeout do pool de conexões
5. Validação de dados: Validar integridade dos dados na camada de modelo ou aplicação

python

 # Gerenciando 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:
    funcionario = Funcionario(nome="Usuário Contexto", email="contexto@empresa.com")
    db.add(funcionario)

Conclusão

O SQLAlchemy ORM oferece uma forma poderosa e flexível de manipulação de banco 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 melhores práticas

O SQLAlchemy possui muitas características avançadas, como propriedades híbridas, escuta de eventos e consultas personalizadas, que valem a pena explorar aprofundadamente.