O SQLAlchemy é o ORM (Object Relational Mapper) mais utilizado no ecossistema Python. Se você trabalha com bancos de dados relacionais, dominar o SQLAlchemy é uma habilidade essencial que separa desenvolvedores iniciantes dos profissionais. Neste guia completo, você aprenderá desde os conceitos fundamentais até técnicas avançadas de modelagem e consulta de dados.
Segundo a documentação oficial do SQLAlchemy, a biblioteca fornece um conjunto completo de padrões para acesso a banco de dados de alto desempenho, adaptados para Python. Seja você um desenvolvedor web, analista de dados ou engenheiro de software, o SQLAlchemy vai transformar a forma como você interage com bancos de dados.
O Que é um ORM e Por Que Usar SQLAlchemy?
Um ORM (Object Relational Mapper) é uma técnica que permite trabalhar com bancos de dados relacionais usando objetos e métodos da sua linguagem de programação, em vez de escrever SQL puro. O SQLAlchemy leva esse conceito a um nível profissional, oferecendo dois modos de uso distintos:
- SQLAlchemy Core: Uma abstração focada em SQL expression language, ideal para quem prefere mais controle sobre as queries
- SQLAlchemy ORM: Uma camada completa de mapeamento objeto-relacional que transforma linhas de tabelas em objetos Python
O tutorial do Real Python sobre SQLAlchemy destaca que a biblioteca é amplamente adotada por empresas como Reddit, Dropbox e Yelp, justamente por sua flexibilidade e desempenho.
Vantagens do SQLAlchemy
- Independência de banco: O mesmo código funciona com SQLite, PostgreSQL, MySQL, Oracle e mais
- Segurança: Proteção nativa contra SQL injection através de queries parametrizadas
- Produtividade: Menos código boilerplate comparado a SQL puro
- Migrações: Integração com Alembic para versionamento de schema
- Performance: Lazy e eager loading otimizam consultas
Instalação e Configuração
A instalação do SQLAlchemy é simples usando pip:
pip install sqlalchemy
Para instalar com suporte a um banco específico:
# Para PostgreSQL
pip install sqlalchemy psycopg2
Para MySQL
pip install sqlalchemy pymysql
Apenas SQLite (já incluso no Python)
pip install sqlalchemy
Você pode verificar a instalação no repositório oficial do SQLAlchemy no PyPI para garantir a versão mais recente.
Conexão com o Banco de Dados
O primeiro passo é criar uma engine de conexão. O SQLAlchemy usa uma string de conexão chamada database URL:
from sqlalchemy import create_engine
SQLite (banco local)
engine = create_engine('sqlite:///meu_banco.db', echo=True)
PostgreSQL
engine = create_engine('postgresql://usuario:senha@localhost:5432/meubanco')
MySQL
engine = create_engine('mysql+pymysql://usuario:senha@localhost:3306/meubanco')
O parâmetro echo=True exibe todas as queries SQL executadas, essencial para debugging e aprendizado.
Definindo Models com Declarative Base
Com o SQLAlchemy ORM, você define suas tabelas como classes Python. A abordagem mais moderna é usar o sistema declarativo:
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
Base = declarative_base()
class Produto(Base):
tablename = 'produtos'
id = Column(Integer, primary_key=True, autoincrement=True)
nome = Column(String(200), nullable=False)
preco = Column(Float, nullable=False)
categoria = Column(String(100))
data_criacao = Column(DateTime, default=datetime.utcnow)
def __repr__(self):
return f"<Produto(nome='{self.nome}', preco={self.preco})>"
Criando as Tabelas
# Criar todas as tabelas definidas
Base.metadata.create_all(engine)
O SQLAlchemy gera automaticamente as instruções CREATE TABLE baseadas nas suas classes. Consulte a documentação sobre tabelas declarativas para mais detalhes.
Sessions: O Coração das Operações
A Session é o mecanismo central para todas as operações de banco no SQLAlchemy ORM. Ela gerencia transações e mantém um cache de objetos:
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()
Sempre use sessions com gerenciamento de contexto para garantir que as conexões sejam fechadas corretamente:
with Session() as session:
# suas operações aqui
pass
Operações CRUD com SQLAlchemy
CREATE — Inserindo Dados
with Session() as session:
produto = Produto(
nome='Notebook Dell XPS',
preco=8999.90,
categoria='Eletrônicos'
)
session.add(produto)
session.commit()
print(f"Produto criado com ID: {produto.id}")
Para inserir múltiplos registros:
produtos = [
Produto(nome='Mouse Gamer', preco=249.90, categoria='Periféricos'),
Produto(nome='Teclado Mecânico', preco=599.90, categoria='Periféricos'),
Produto(nome='Monitor 27" 4K', preco=3499.90, categoria='Monitores'),
]
session.add_all(produtos)
session.commit()
READ — Consultando Dados
O SQLAlchemy oferece uma API rica para consultas. Veja os padrões mais comuns:
# Buscar todos os registros
todos_produtos = session.query(Produto).all()
Buscar por ID
produto = session.query(Produto).get(1)
Primeiro registro que atende ao filtro
produto = session.query(Produto).filter_by(nome='Mouse Gamer').first()
Filtrar com condições
produtos_caros = session.query(Produto).filter(
Produto.preco >= 1000
).all()
UPDATE — Atualizando Dados
with Session() as session:
produto = session.query(Produto).get(1)
produto.preco = 8499.90
produto.nome = 'Notebook Dell XPS 15'
session.commit()
DELETE — Removendo Dados
with Session() as session:
produto = session.query(Produto).get(2)
session.delete(produto)
session.commit()
Relacionamentos Entre Tabelas
Um dos recursos mais poderosos do SQLAlchemy é o gerenciamento de relacionamentos entre tabelas. Vamos criar um exemplo completo com clientes e pedidos.
One-to-Many (Um para Muitos)
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
class Cliente(Base):
tablename = 'clientes'
id = Column(Integer, primary_key=True)
nome = Column(String(100), nullable=False)
email = Column(String(200), unique=True, nullable=False)
pedidos = relationship('Pedido', back_populates='cliente')
class Pedido(Base):
tablename = 'pedidos'
id = Column(Integer, primary_key=True)
data = Column(DateTime, default=datetime.utcnow)
valor_total = Column(Float)
cliente_id = Column(Integer, ForeignKey('clientes.id'))
cliente = relationship('Cliente', back_populates='pedidos')
Agora você pode navegar pelos relacionamentos naturalmente:
# Acessar pedidos de um cliente
cliente = session.query(Cliente).get(1)
for pedido in cliente.pedidos:
print(f"Pedido #{pedido.id} - R$ {pedido.valor_total}")
Acessar cliente de um pedido
pedido = session.query(Pedido).get(1)
print(f"Cliente: {pedido.cliente.nome}")
Many-to-Many (Muitos para Muitos)
Para relacionamentos muitos-para-muitos, usamos uma tabela de associação:
from sqlalchemy import Table, Text
Tabela de associação
tabela_pedido_produto = Table(
'pedido_produtos', Base.metadata,
Column('pedido_id', Integer, ForeignKey('pedidos.id')),
Column('produto_id', Integer, ForeignKey('produtos.id')),
)
class Produto(Base):
tablename = 'produtos'
id = Column(Integer, primary_key=True)
nome = Column(String(200))
preco = Column(Float)
descricao = Column(Text)
pedidos = relationship('Pedido', secondary=tabela_pedido_produto, back_populates='produtos')
class Pedido(Base):
tablename = 'pedidos'
id = Column(Integer, primary_key=True)
data = Column(DateTime, default=datetime.utcnow)
valor_total = Column(Float)
cliente_id = Column(Integer, ForeignKey('clientes.id'))
cliente = relationship('Cliente', back_populates='pedidos')
produtos = relationship('Produto', secondary=tabela_pedido_produto, back_populates='pedidos')
Para mais informações sobre relacionamentos, acesse a documentação oficial de relacionamentos do SQLAlchemy.
Consultas Avançadas
Filtros Complexos
from sqlalchemy import and_, or_, not_
Usar AND, OR, NOT
produtos = session.query(Produto).filter(
and(
Produto.preco >= 100,
Produto.preco <= 5000,
or(
Produto.categoria == 'Eletrônicos',
Produto.categoria == 'Periféricos'
)
)
).all()
Joins Explicitos
# JOIN entre tabelas
resultados = session.query(Pedido, Cliente).join(
Cliente, Pedido.cliente_id == Cliente.id
).all()
for pedido, cliente in resultados:
print(f"Pedido {pedido.id} - {cliente.nome} - R$ {pedido.valor_total}")
Agregações e Group By
from sqlalchemy import func
Total de vendas por cliente
resultados = session.query(
Cliente.nome,
func.count(Pedido.id).label('total_pedidos'),
func.sum(Pedido.valor_total).label('valor_total')
).join(Pedido).group_by(Cliente.id).all()
for nome, total, valor in resultados:
print(f"{nome}: {total} pedidos - R$ {valor:.2f}")
O SQLAlchemy suporta todas as funções SQL padrão através de func. Consulte a documentação de funções SQL do SQLAlchemy para a lista completa.
Subqueries
# Subquery: clientes com pedidos acima da média
subquery = session.query(
func.avg(Pedido.valor_total)
).scalar_subquery()
clientes_acima = session.query(Cliente).join(Pedido).filter(
Pedido.valor_total > subquery
).distinct().all()
Lazy Loading vs Eager Loading
Um conceito crucial para performance é entender quando e como os dados relacionados são carregados:
- Lazy Loading (padrão): Os relacionamentos são carregados apenas quando acessados. Pode causar N+1 queries se usado sem cuidado
- Eager Loading: Carrega tudo em uma única query usando
joinedloadousubqueryload
from sqlalchemy.orm import joinedload
Eager loading com JOIN
clientes = session.query(Cliente).options(
joinedload(Cliente.pedidos)
).all()
Agora todos os pedidos são carregados na mesma query
for cliente in clientes:
print(f"{cliente.nome}: {len(cliente.pedidos)} pedidos")
Migrações com Alembic
O Alembic é a ferramenta de migração oficial do SQLAlchemy, criada pelo mesmo autor. Ela permite versionar o schema do banco de dados:
pip install alembic
alembic init alembic
Configure o arquivo alembic.ini com a URL do seu banco e comece a criar migrações:
# Criar uma nova migração automática
alembic revision --autogenerate -m "criar_tabela_produtos"
Aplicar a migração
alembic upgrade head
Para um guia completo, visite a documentação oficial do Alembic.
Boas Práticas com SQLAlchemy
1. Sempre Use Context Manager para Sessions
# Correto
with Session() as session:
session.add(produto)
session.commit()
Incorreto (pode vazar conexão)
session = Session()
session.add(produto)
session.commit()
2. Índices para Performance
class Produto(Base):
__tablename__ = 'produtos'
id = Column(Integer, primary_key=True)
nome = Column(String(200), index=True) # Índice para buscas por nome
categoria = Column(String(100), index=True)
3. Use Enum para Campos Fixos
from enum import Enum as PyEnum
from sqlalchemy import Enum as SAEnum
class StatusPedido(PyEnum):
PENDENTE = 'pendente'
PAGO = 'pago'
ENVIADO = 'enviado'
ENTREGUE = 'entregue'
class Pedido(Base):
tablename = 'pedidos'
status = Column(SAEnum(StatusPedido), default=StatusPedido.PENDENTE)
4. Evite N+1 Queries
O problema N+1 ocorre quando você carrega N registros e depois faz mais N consultas para cada relacionamento. Use joinedload ou subqueryload para evitar isso.
SQLAlchemy com FastAPI
Integrar SQLAlchemy com FastAPI é uma combinação poderosa para criar APIs REST de alta performance. A estrutura recomendada usa dependências para gerenciar sessions:
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
app = FastAPI()
def get_db():
db = Session()
try:
yield db
finally:
db.close()
@app.get("/produtos/")
def listar_produtos(db: Session = Depends(get_db)):
produtos = db.query(Produto).all()
return produtos
Para se aprofundar na criação de APIs, veja nosso guia de FastAPI Python: Criando APIs RESTful.
SQLAlchemy vs Outras Ferramentas
| Característica | SQLAlchemy | Django ORM | Peewee |
|---|---|---|---|
| Flexibilidade | Alta (Core + ORM) | Média | Média |
| Curva de aprendizado | íngreme | Moderada | Suave |
| Suporte a bancos | SQLite, PG, MySQL, Oracle, MSSQL | SQLite, PG, MySQL, Oracle | SQLite, PG, MySQL |
| Performance | Excelente | Boa | Boa |
| Migrações | Alembic | built-in | Playhouse |
| Async support | Sim (asyncio) | Sim (3.1+) | Limitado |
Suporte a Async com SQLAlchemy
A partir da versão 1.4, o SQLAlchemy oferece suporte nativo a async/await, permitindo operações não-bloqueantes com bancos de dados:
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
import asyncio
async_engine = create_async_engine('sqlite+aiosqlite:///meu_banco.db')
AsyncSessionLocal = sessionmaker(
asyncengine, class=AsyncSession
)
async def get_produtos():
async with AsyncSessionLocal() as session:
result = await session.execute('SELECT * FROM produtos')
return result.fetchall()
asyncio.run(get_produtos())
A programação assíncrona é um tópico avançado que vale a pena estudar. Confira nosso guia de Python Async/Await: Programação Assíncrona.
Conclusão
O SQLAlchemy é muito mais que um simples ORM — é um ecossistema completo para trabalhar com bancos de dados relacionais em Python. Desde pequenos projetos com SQLite até sistemas corporativos com PostgreSQL, a biblioteca oferece as ferramentas necessárias para construir camadas de dados robustas, seguras e performáticas.
O que vimos neste guia:
- Instalação e configuração com diferentes bancos de dados
- Definição de models usando a abordagem declarativa
- Operações CRUD completas
- Relacionamentos one-to-many e many-to-many
- Consultas avançadas com joins, subqueries e agregações
- Estratégias de carregamento (lazy vs eager)
- Migrações com Alembic
- Integração com FastAPI
- Suporte a async
Para continuar seus estudos, recomendo fortemente a leitura da documentação oficial do SQLAlchemy 2.0, que é uma das melhores documentações técnicas do ecossistema Python. Pratique criando modelos, experimente com diferentes tipos de consultas e, principalmente, construa projetos reais.
O SQLAlchemy também se integra perfeitamente com ferramentas como Pytest para testes automatizados e mypy para verificação de tipos, garantindo código ainda mais robusto.
Comece hoje mesmo a usar o SQLAlchemy nos seus projetos Python e eleve a qualidade da sua camada de dados ao nível profissional!