Capítulo 07: Relacionamentos entre Tabelas (Relationship) e Consultas com Join
Especialização em Backend com Python & FastAPI • FastAPI & Python 3.12+ • Pydantic v2, SQLModel, Async/Await e Microsserviços
🗺️ Mapa Conceitual do Tópico
flowchart TD
A["Tabela Relacional 1: Time"] -->|"1 : N"| B["Tabela Relacional 2: Heroi"]
C["Chave Estrangeira: foreign_key='time.id'"] --> B
D["Atributo ORM: Relationship(back_populates='herois')"] -.-> A
E["Atributo ORM: Relationship(back_populates='time')"] -.-> B
subgraph PROB_N1["O Problema das N+1 Consultas em Async"]
F["1 Query Inicial: SELECT * FROM time LIMIT 50"]
G["+ 50 Queries Implícitas (Lazy Loading): SELECT FROM heroi WHERE time_id = ?"]
H["💥 Erro em AsyncSession: MissingGreenlet (Lazy loading síncrono é proibido)"]
F --> G --> H
end
subgraph SOLUCAO["Solução Canônica: Eager Loading Assíncrono"]
I["select(Time).options(selectinload(Time.herois))"]
J["Execução: Exatamente 2 Queries Otimizadas via IN (...)"]
I --> J
end
style A fill:#e1f5fe,stroke:#03a9f4,stroke-width:2px
style B fill:#fff3e0,stroke:#ff9800,stroke-width:2px
style C fill:#ede7f6,stroke:#7e57c2,stroke-width:2px
style D fill:#e8f5e9,stroke:#4caf50,stroke-width:2px
style E fill:#fce4ec,stroke:#e91e63,stroke-width:2px
style F fill:#ffebee,stroke:#e53935,stroke-width:2px
style G fill:#ffebee,stroke:#e53935,stroke-width:2px
style H fill:#b71c1c,stroke:#b71c1c,stroke-width:2px,color:#fff
style I fill:#e8f5e9,stroke:#4caf50,stroke-width:2px
style J fill:#e0f2f1,stroke:#009688,stroke-width:2px
🏛️ 1. Chave Estrangeira (foreign_key) vs Associação ORM (Relationship)
No desenvolvimento com SQLModel, desenvolvedores frequentemente confundem a chave estrangeira relacional com a propriedade de relacionamento do ORM. Eles operam em camadas conceituais completamente distintas:
Field(foreign_key="tabela.coluna"): É uma definição puramente de Banco de Dados. Gera uma restrição DDL de integridade referencial SQL (CONSTRAINT fk_... FOREIGN KEY). Ela impede fisicamente que um registro aponte para um ID inexistente no banco.Relationship(back_populates="campo"): É uma conveniência puramente de Software / ORM. Não cria coluna no banco. Ele permite navegar entre instâncias em Python utilizando notação de ponto (time.heroisouheroi.time), mantendo ambas as pontas sincronizadas automaticamente na sessão ativa.
Mapeamento 1:N com SQLModel Idiomático
from typing import Optional, List
from sqlmodel import SQLModel, Field, Relationship
class Time(SQLModel, table=True):
__tablename__ = "times"
id: Optional[int] = Field(default=None, primary_key=True)
nome: str = Field(index=True, max_length=100)
quartel_general: str
# Lado "Um": coleciona múltiplos heróis (1:N)
herois: List["Heroi"] = Relationship(back_populates="time")
class Heroi(SQLModel, table=True):
__tablename__ = "herois"
id: Optional[int] = Field(default=None, primary_key=True)
nome: str = Field(index=True)
nome_secreto: str
# Coluna física no banco de dados
time_id: Optional[int] = Field(default=None, foreign_key="times.id")
# Lado "Muitos": navega até o time pai
time: Optional[Time] = Relationship(back_populates="herois")
🔗 2. Relacionamento Muitos-para-Muitos (N:N) com link_model
Em cenários N:N (por exemplo, um Heroi pode pertencer a múltiplas Equipes, e uma Equipe é composta por múltiplos Heróis), é obrigatório criar uma Tabela de Junção Associativa (Link Table). No SQLModel, isso é feito de forma elegante via link_model:
class HeroiEquipeLink(SQLModel, table=True):
__tablename__ = "heroi_equipe_links"
heroi_id: Optional[int] = Field(
default=None, foreign_key="herois.id", primary_key=True
)
equipe_id: Optional[int] = Field(
default=None, foreign_key="equipes.id", primary_key=True
)
class Equipe(SQLModel, table=True):
__tablename__ = "equipes"
id: Optional[int] = Field(default=None, primary_key=True)
nome: str = Field(unique=True)
# Associação N:N mapeada via link_model
membros: List[Heroi] = Relationship(
back_populates="equipes",
link_model=HeroiEquipeLink
)
⚡ 3. O Problema das N+1 Consultas e o Erro MissingGreenlet
Em ORMs síncronos tradicionais, o padrão padrão para acessar relacionamentos é o Lazy Loading (carregamento sob demanda): quando o código lê time.herois, o SQLAlchemy executa silenciosamente um SELECT síncrono no banco.
No entanto, em um ambiente FastAPI Assíncrono (AsyncSession), o Lazy Loading síncrono é expressamente proibido e disparará uma falha catastrófica:
sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called;
can't call __init__ and navigate attributes under async mode
Para solucionar esse problema e simultaneamente eliminar o gargalo de performance de disparar N+1 queries, utilizamos estratégias de Eager Loading explícito:
| Estratégia de Carregamento | Função SQLAlchemy | Comportamento SQL Gerado | Quando Utilizar |
|---|---|---|---|
selectinload |
options(selectinload(...)) |
Emite 2 queries: a principal e uma com WHERE id IN (...) |
Recomendado para coleções 1:N e N:N. Não duplica dados na rede. |
joinedload |
options(joinedload(...)) |
Emite 1 query única usando LEFT OUTER JOIN |
Recomendado para relacionamentos N:1 e 1:1 (lado filho para pai). |
join Direto |
select().join(...) |
Cláusula INNER JOIN clássica |
Quando você quer filtrar ou projetar colunas tabulares agregadas. |
from sqlmodel import select
from sqlalchemy.orm import selectinload, joinedload
from sqlmodel.ext.asyncio.session import AsyncSession
# Exemplo de consulta otimizada em 2 etapas com selectinload
async def buscar_times_com_seus_herois(session: AsyncSession) -> list[Time]:
stmt = (
select(Time)
.options(selectinload(Time.herois))
.order_by(Time.nome)
)
resultado = await session.exec(stmt)
return list(resultado.all())
🔄 4. Prevenção de Loops Infinitos de Serialização no Pydantic
Ao retornar entidades com relacionamentos bidirecionais (Time -> herois -> time -> herois...), a serialização JSON padrão do Pydantic entra em um loop infinito de recursão (RecursionError). A solução canônica é definir Schemas DTO de Resposta Específicos:
# Schemas simples sem aninhamento
class HeroiRead(SQLModel):
id: int
nome: str
nome_secreto: str
time_id: Optional[int] = None
class TimeRead(SQLModel):
id: int
nome: str
quartel_general: str
# Schema estendido que incorpora a coleção de filhos
class TimeReadWithHerois(TimeRead):
herois: list[HeroiRead] = []
# No endpoint FastAPI:
@app.get("/times/{time_id}", response_model=TimeReadWithHerois)
async def obter_time(time_id: int, session: SessionDep):
stmt = select(Time).where(Time.id == time_id).options(selectinload(Time.herois))
resultado = await session.exec(stmt)
time = resultado.first()
if not time:
raise HTTPException(status_code=404, detail="Time não localizado")
return time