Ficha 02 · ORM SQLAlchemy
- Definir modelos
- Fazer CRUD
- Resolver N+1
- Aplicar migrações
Parte I · Modelos
Exercício 1 · Definir modelos (20 pts)
Define modelos SQLAlchemy para uma escola: Aluno, Disciplina, Inscricao (N:M com nota):
from sqlalchemy import Column, Integer, String, ForeignKey, Float, DateTime
from sqlalchemy.orm import declarative_base, relationship
from datetime import datetime
Base = declarative_base()
class Aluno(Base):
__tablename__ = "aluno"
id = Column(Integer, primary_key=True)
numero = Column(String(10), unique=True, nullable=False)
nome = Column(String(150), nullable=False)
email = Column(String(150), unique=True)
inscricoes = relationship("Inscricao", back_populates="aluno",
cascade="all, delete-orphan")
def __repr__(self):
return f"<Aluno(nº={self.numero}, nome={self.nome!r})>"
class Disciplina(Base):
__tablename__ = "disciplina"
id = Column(Integer, primary_key=True)
codigo = Column(String(20), unique=True, nullable=False)
nome = Column(String(100), nullable=False)
inscricoes = relationship("Inscricao", back_populates="disciplina")
class Inscricao(Base):
__tablename__ = "inscricao"
aluno_id = Column(Integer, ForeignKey("aluno.id"), primary_key=True)
disciplina_id = Column(Integer, ForeignKey("disciplina.id"), primary_key=True)
ano_lectivo = Column(String(9), primary_key=True) # ex.: "2026/2027"
nota = Column(Float)
criada = Column(DateTime, default=datetime.utcnow)
aluno = relationship("Aluno", back_populates="inscricoes")
disciplina = relationship("Disciplina", back_populates="inscricoes")
PK composta inclui ano_lectivo para permitir repetição da disciplina.
Exercício 2 · CRUD (15 pts)
Escreve código para:
a) Criar 2 alunos. b) Criar 1 disciplina. c) Inscrever ambos os alunos na disciplina. d) Listar todas as inscrições com nome do aluno + disciplina.
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine("sqlite:///escola.db")
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
# a)
ana = Aluno(numero="A001", nome="Ana Silva", email="ana@ex.pt")
joao = Aluno(numero="A002", nome="João Costa", email="joao@ex.pt")
session.add_all([ana, joao])
session.commit()
# b)
prog = Disciplina(codigo="PROG", nome="Programação")
session.add(prog)
session.commit()
# c)
i1 = Inscricao(aluno_id=ana.id, disciplina_id=prog.id, ano_lectivo="2026/2027")
i2 = Inscricao(aluno_id=joao.id, disciplina_id=prog.id, ano_lectivo="2026/2027")
session.add_all([i1, i2])
session.commit()
# d)
inscricoes = (session.query(Inscricao, Aluno, Disciplina)
.join(Aluno, Inscricao.aluno_id == Aluno.id)
.join(Disciplina, Inscricao.disciplina_id == Disciplina.id)
.all())
for ins, al, dis in inscricoes:
print(f"{al.nome} — {dis.nome} ({ins.ano_lectivo})")
Parte II · Queries
Exercício 3 · Filtros (15 pts)
Para os modelos do exercício 1, escreve queries SQLAlchemy para:
a) Alunos cujo nome começa por "A".
b) Disciplinas com pelo menos 1 inscrição em 2026/2027.
c) Top 5 alunos com mais inscrições.
from sqlalchemy import func, distinct
# a)
session.query(Aluno).filter(Aluno.nome.like("A%")).all()
# b)
disciplinas = (session.query(Disciplina)
.join(Inscricao)
.filter(Inscricao.ano_lectivo == "2026/2027")
.distinct()
.all())
# c)
top5 = (session.query(Aluno, func.count(Inscricao.disciplina_id).label("n"))
.join(Inscricao)
.group_by(Aluno.id)
.order_by(func.count(Inscricao.disciplina_id).desc())
.limit(5)
.all())
for aluno, n in top5:
print(f"{aluno.nome}: {n} inscrições")
Exercício 4 · N+1 problem (20 pts)
a) Identifica o problema neste código:
alunos = session.query(Aluno).all()
for a in alunos:
print(a.nome, len(a.inscricoes))
b) Reescreve para evitar o problema.
c) Quantas queries de BD em cada versão (assumindo 100 alunos)?
a) Problema: N+1 queries. A primeira query(Aluno).all() faz 1 query. Mas a.inscricoes é lazy load — cada len() dispara nova query para buscar inscrições daquele aluno.
b) Eager loading:
from sqlalchemy.orm import joinedload
alunos = session.query(Aluno).options(
joinedload(Aluno.inscricoes)
).all()
for a in alunos:
print(a.nome, len(a.inscricoes))
Ou com selectinload (uma query separada com IN, geralmente mais eficiente para colecções):
from sqlalchemy.orm import selectinload
alunos = session.query(Aluno).options(
selectinload(Aluno.inscricoes)
).all()
c) Comparação:
| Versão | Queries (100 alunos) |
|---|---|
| Versão a (lazy) | 1 + 100 = 101 queries |
joinedload |
1 query com JOIN |
selectinload |
2 queries (1 para alunos, 1 com IN para inscrições) |
Em produção, N+1 é fonte comum de problemas de performance.
Parte III · Migrações
Exercício 5 · Alembic workflow (10 pts)
Descreve passo-a-passo como adicionar coluna telefone à tabela Aluno usando Alembic:
# 1. Adicionar coluna no modelo Python
# (em models.py)
# telefone = Column(String(20))
# 2. Gerar migração automaticamente
alembic revision --autogenerate -m "Add telefone to Aluno"
# 3. Verificar ficheiro de migração gerado em migrations/versions/
# Conferir que upgrade() faz add_column e downgrade() faz drop_column
# 4. Aplicar à BD
alembic upgrade head
# 5. Verificar:
alembic current
# 6. Commit do ficheiro de migração ao Git
git add migrations/versions/abc123_add_telefone_to_aluno.py
git commit -m "feat: add telefone column"
# 7. Equipa, ao puxar, faz:
git pull
alembic upgrade head
Reverter (caso erro):
alembic downgrade -1
Parte IV · Cenário
Exercício 6 · Repository com SQLAlchemy (20 pts)
Reescreve o ClienteRepository da ficha anterior, mas agora usando SQLAlchemy ORM.
from sqlalchemy.orm import Session
from sqlalchemy.exc import NoResultFound, IntegrityError
class ClienteRepository:
def __init__(self, session: Session):
self.session = session
def get(self, id: int) -> "Cliente | None":
return self.session.get(Cliente, id)
def get_by_email(self, email: str) -> "Cliente | None":
return (self.session.query(Cliente)
.filter_by(email=email)
.first())
def list_all(self) -> list["Cliente"]:
return (self.session.query(Cliente)
.order_by(Cliente.nome)
.all())
def create(self, nome: str, email: str) -> Cliente:
c = Cliente(nome=nome, email=email)
self.session.add(c)
try:
self.session.commit()
except IntegrityError:
self.session.rollback()
raise ValueError("Email já existe") from None
return c
def update(self, id: int, **kwargs) -> Cliente:
c = self.session.get(Cliente, id)
if c is None:
raise ValueError(f"Cliente {id} não existe")
allowed = {"nome", "email"}
for k, v in kwargs.items():
if k not in allowed:
raise ValueError(f"Coluna não permitida: {k}")
setattr(c, k, v)
self.session.commit()
return c
def delete(self, id: int) -> None:
c = self.session.get(Cliente, id)
if c is not None:
self.session.delete(c)
self.session.commit()
# Uso
session = Session(engine)
repo = ClienteRepository(session)
ana = repo.create("Ana", "ana@ex.pt")
repo.update(ana.id, nome="Ana Silva")
print(repo.get(ana.id).nome)
Vantagens vs SQL directo:
- Validação automática de tipos.
- Type hints completas.
- Excepções claras (IntegrityError).
- Cleaner code.