O que é SQLAlchemy ORM e por que você precisa dela
SQLAlchemy é a biblioteca mais madura e poderosa para trabalhar com bancos de dados em Python. A sigla ORM significa Object-Relational Mapping, ou seja, ela mapeia objetos Python diretamente para tabelas do banco de dados. Em vez de escrever SQL puro, você trabalha com classes Python que representam suas entidades, e a biblioteca cuida da tradução para SQL.
A grande vantagem é que você ganha abstração e segurança. Não precisa se preocupar com sintaxe SQL específica de cada banco de dados, não está vulnerável a SQL injection, e o código fica mais legível e manutenível. Se você precisar mudar de PostgreSQL para MySQL, basta alterar a string de conexão — seu código Python permanece praticamente igual.
Models: Definindo suas Entidades
Estrutura Básica de um Model
Um Model é uma classe Python que representa uma tabela no banco de dados. Cada atributo da classe é uma coluna, e cada instância é um registro. Você define um Model herdando de uma classe base que SQLAlchemy fornece.
from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
# Criamos a base que todas as nossas classes vão herdar
Base = declarative_base()
class Usuario(Base):
__tablename__ = 'usuarios'
id = Column(Integer, primary_key=True)
nome = Column(String(100), nullable=False)
email = Column(String(120), unique=True, nullable=False)
data_criacao = Column(DateTime, default=datetime.utcnow)
# Criamos o engine (conexão com o banco)
engine = create_engine('sqlite:///banco.db')
# Criamos as tabelas no banco
Base.metadata.create_all(engine)
O __tablename__ define o nome da tabela no banco de dados. O Column especifica tipo, restrições (primary_key, nullable, unique). O engine é responsável por gerenciar a conexão com o banco — neste exemplo, usamos SQLite por simplicidade, mas você pode usar PostgreSQL, MySQL, Oracle, etc.
Tipos de Dados e Validações
SQLAlchemy oferece tipos de dados robustos que mapeiam para tipos nativos do banco. Além disso, você pode adicionar validações simples direto nas colunas.
from sqlalchemy import Boolean, Float, Text, Enum
import enum
class StatusPedido(enum.Enum):
PENDENTE = "pendente"
PROCESSANDO = "processando"
CONCLUIDO = "concluido"
class Pedido(Base):
__tablename__ = 'pedidos'
id = Column(Integer, primary_key=True)
descricao = Column(Text) # Texto longo
valor = Column(Float, nullable=False) # Números decimais
ativo = Column(Boolean, default=True) # Booleanos
status = Column(Enum(StatusPedido), default=StatusPedido.PENDENTE)
quantidade = Column(Integer, default=1)
Cada tipo de coluna tem seu mapeamento automático para o banco. Float vira DECIMAL, Boolean vira um campo booleano (ou INTEGER em alguns bancos), Enum permite usar enumerações Python de forma segura.
Relacionamentos: Conectando Models
One-to-Many (Um para Muitos)
Este é o relacionamento mais comum. Um usuário pode ter muitos pedidos, mas cada pedido pertence a apenas um usuário. Você define isso com ForeignKey e relationship.
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
class Usuario(Base):
__tablename__ = 'usuarios'
id = Column(Integer, primary_key=True)
nome = Column(String(100), nullable=False)
# Aqui vinculamos os pedidos do usuário
pedidos = relationship('Pedido', back_populates='usuario')
class Pedido(Base):
__tablename__ = 'pedidos'
id = Column(Integer, primary_key=True)
descricao = Column(String(255), nullable=False)
usuario_id = Column(Integer, ForeignKey('usuarios.id'), nullable=False)
# Aqui referenciamos o usuário
usuario = relationship('Usuario', back_populates='pedidos')
O ForeignKey garante integridade referencial no banco de dados. O relationship é um atributo Python que permite acessar os dados relacionados facilmente. O back_populates cria um link bidirecional — você acessa usuario.pedidos ou pedido.usuario automaticamente.
Many-to-Many (Muitos para Muitos)
Quando duas entidades se relacionam de forma bidirecional sem hierarquia, usamos tabelas de associação. Por exemplo, alunos e cursos: um aluno pode estar em vários cursos, e um curso pode ter vários alunos.
from sqlalchemy import Table
# Tabela de associação (sem classe correspondente)
aluno_curso = Table(
'aluno_curso',
Base.metadata,
Column('aluno_id', Integer, ForeignKey('alunos.id'), primary_key=True),
Column('curso_id', Integer, ForeignKey('cursos.id'), primary_key=True)
)
class Aluno(Base):
__tablename__ = 'alunos'
id = Column(Integer, primary_key=True)
nome = Column(String(100), nullable=False)
cursos = relationship('Curso', secondary=aluno_curso, back_populates='alunos')
class Curso(Base):
__tablename__ = 'cursos'
id = Column(Integer, primary_key=True)
titulo = Column(String(150), nullable=False)
alunos = relationship('Aluno', secondary=aluno_curso, back_populates='cursos')
O argumento secondary aponta para a tabela de junção. Isso permite acesso simples: aluno.cursos retorna uma lista de cursos, sem você precisar escrever queries manualmente.
One-to-One (Um para Um)
Menos comum, mas útil quando uma entidade tem exatamente um relacionamento com outra. Usamos uselist=False para isso.
class Pessoa(Base):
__tablename__ = 'pessoas'
id = Column(Integer, primary_key=True)
nome = Column(String(100), nullable=False)
passaporte = relationship('Passaporte', uselist=False, back_populates='pessoa')
class Passaporte(Base):
__tablename__ = 'passaportes'
id = Column(Integer, primary_key=True)
numero = Column(String(20), unique=True, nullable=False)
pessoa_id = Column(Integer, ForeignKey('pessoas.id'), unique=True)
pessoa = relationship('Pessoa', back_populates='passaporte')
O uselist=False faz com que pessoa.passaporte retorne um objeto único, não uma lista.
Sessions: Gerenciando o Ciclo de Vida dos Objetos
O que é uma Session
A Session é fundamental em SQLAlchemy ORM. Ela funciona como um gerenciador de contexto que rastreia mudanças nos objetos e sincroniza com o banco de dados. Você não trabalha diretamente com o banco — tudo passa pela Session.
from sqlalchemy.orm import sessionmaker
# Cria a fábrica de sessions
SessionLocal = sessionmaker(bind=engine)
# Cria uma nova session
session = SessionLocal()
try:
# Cria um novo usuário
novo_usuario = Usuario(nome='João Silva', email='joao@example.com')
# Adiciona à session (mas ainda não está no banco)
session.add(novo_usuario)
# Commita as mudanças (agora vai para o banco)
session.commit()
print(f"Usuário criado com ID: {novo_usuario.id}")
finally:
# Sempre fecha a session
session.close()
Entenda o fluxo: você cria objetos Python, adiciona à session, e faz commit para persisti-los no banco. A session funciona como um buffer — você pode fazer múltiplas operações e confirmar tudo de uma vez.
Operações CRUD
CRUD significa Create, Read, Update, Delete. Vamos cobrir cada uma com exemplos práticos.
from sqlalchemy.orm import sessionmaker
SessionLocal = sessionmaker(bind=engine)
# CREATE
def criar_usuario(nome, email):
session = SessionLocal()
try:
usuario = Usuario(nome=nome, email=email)
session.add(usuario)
session.commit()
return usuario
except Exception as e:
session.rollback()
print(f"Erro ao criar: {e}")
finally:
session.close()
# READ - Buscar um por ID
def obter_usuario_por_id(user_id):
session = SessionLocal()
try:
usuario = session.query(Usuario).filter(Usuario.id == user_id).first()
return usuario
finally:
session.close()
# READ - Buscar vários
def listar_usuarios():
session = SessionLocal()
try:
usuarios = session.query(Usuario).all()
return usuarios
finally:
session.close()
# UPDATE
def atualizar_usuario(user_id, novo_nome):
session = SessionLocal()
try:
usuario = session.query(Usuario).filter(Usuario.id == user_id).first()
if usuario:
usuario.nome = novo_nome
session.commit()
return usuario
except Exception as e:
session.rollback()
print(f"Erro ao atualizar: {e}")
finally:
session.close()
# DELETE
def deletar_usuario(user_id):
session = SessionLocal()
try:
usuario = session.query(Usuario).filter(Usuario.id == user_id).first()
if usuario:
session.delete(usuario)
session.commit()
except Exception as e:
session.rollback()
print(f"Erro ao deletar: {e}")
finally:
session.close()
Note que sempre usamos try/finally para garantir que a session seja fechada, mesmo com erros. O rollback() desfaz mudanças não commitadas — é essencial para manter consistência.
Queries Avançadas e Filtering
SQLAlchemy permite queries expresivas que são traduzidas para SQL.
from sqlalchemy import and_, or_
session = SessionLocal()
# Filtro simples
usuarios = session.query(Usuario).filter(Usuario.nome == 'João Silva').all()
# Múltiplos filtros (AND implícito)
usuarios = session.query(Usuario).filter(
Usuario.nome == 'João Silva',
Usuario.email.like('%@gmail.com')
).all()
# OR
usuarios = session.query(Usuario).filter(
or_(
Usuario.nome == 'João',
Usuario.nome == 'Maria'
)
).all()
# Ordenação
usuarios = session.query(Usuario).order_by(Usuario.nome).all()
# Limite
primeiros_5 = session.query(Usuario).limit(5).all()
# Contar
total = session.query(Usuario).count()
# Acessando relacionamentos
usuario = session.query(Usuario).filter(Usuario.id == 1).first()
if usuario:
# Acessa todos os pedidos do usuário
for pedido in usuario.pedidos:
print(pedido.descricao)
session.close()
O método filter() aceita condições Python que são automaticamente traduzidas para SQL. Operadores como ==, .like(), .in_() funcionam intuitivamente.
Lazy Loading vs Eager Loading
Por padrão, SQLAlchemy carrega relacionamentos sob demanda (lazy loading). Isso pode causar múltiplas queries. Eager loading carrega tudo de uma vez.
from sqlalchemy.orm import joinedload
session = SessionLocal()
# Lazy loading (padrão) - vai fazer uma query para cada pedido acessado
usuarios = session.query(Usuario).all()
for usuario in usuarios:
print(usuario.pedidos) # Cada acesso gera uma query
# Eager loading - carrega tudo em uma única query
usuarios = session.query(Usuario).options(joinedload(Usuario.pedidos)).all()
for usuario in usuarios:
print(usuario.pedidos) # Já está em memória, sem query extra
session.close()
Eager loading é mais eficiente quando você sabe que vai precisar dos dados relacionados. Use .options(joinedload(...)) para carregá-los junto.
Casos de Uso Prático e Padrões
Integração com FastAPI
Um padrão comum é usar SQLAlchemy com FastAPI. Aqui está uma aplicação funcional completa.
from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel
# Setup
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# Model
class Produto(Base):
__tablename__ = "produtos"
id = Column(Integer, primary_key=True)
nome = Column(String(100), nullable=False)
preco = Column(Integer, nullable=False)
Base.metadata.create_all(bind=engine)
# Schemas Pydantic
class ProdutoSchema(BaseModel):
id: int
nome: str
preco: int
class Config:
from_attributes = True
class ProdutoCreate(BaseModel):
nome: str
preco: int
# Dependency para injetar session
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# API
app = FastAPI()
@app.post("/produtos/", response_model=ProdutoSchema)
def criar_produto(produto: ProdutoCreate, db: Session = Depends(get_db)):
novo = Produto(nome=produto.nome, preco=produto.preco)
db.add(novo)
db.commit()
db.refresh(novo)
return novo
@app.get("/produtos/{produto_id}", response_model=ProdutoSchema)
def obter_produto(produto_id: int, db: Session = Depends(get_db)):
return db.query(Produto).filter(Produto.id == produto_id).first()
@app.get("/produtos/", response_model=list[ProdutoSchema])
def listar_produtos(db: Session = Depends(get_db)):
return db.query(Produto).all()
O padrão de dependency injection (Depends(get_db)) fornece uma session limpa para cada requisição e a fecha automaticamente.
Tratamento de Erros e Validação
Erros de banco de dados devem ser tratados com cuidado. SQLAlchemy lança exceções que você deve antecipar.
from sqlalchemy.exc import IntegrityError, SQLAlchemyError
session = SessionLocal()
try:
# Email duplicado vai gerar IntegrityError
usuario1 = Usuario(nome='Ana', email='ana@test.com')
usuario2 = Usuario(nome='Bruno', email='ana@test.com') # Mesmo email
session.add(usuario1)
session.add(usuario2)
session.commit()
except IntegrityError as e:
session.rollback()
print("Email já existe no banco!")
except SQLAlchemyError as e:
session.rollback()
print(f"Erro genérico de banco: {e}")
finally:
session.close()
IntegrityError captura violações de constraints (unique, foreign key, etc). SQLAlchemyError é a classe mãe de todos os erros SQLAlchemy.
Conclusão
Os três pilares que você deve levar deste artigo são:
-
Models são sua ponte com o banco de dados — cada classe herda de Base e mapeia para uma tabela. Defina tipos corretos, restrições claras, e os relacionamentos com precisão. Isso previne bugs e garante integridade dos dados.
-
Relacionamentos exigem planejamento — One-to-Many, Many-to-Many, One-to-One têm usos distintos. Entenda quando usar cada um, porque isso afeta como você consulta os dados e como o banco os organiza. Use
back_populatespara manter referências bidirecionais limpas. -
Sessions são o coração da ORM — toda interação com o banco passa por ela. Sempre feche sessions em finally, use commit() para persisti-las, e rollback() para desfazer mudanças. Eager loading com joinedload previne queries N+1 desnecessárias.