💻 Exemplos: Capítulo 07: Relacionamentos entre Tabelas (Relationship) e Consultas com Join
Especialização em Backend com Python & FastAPI • Demonstrações Práticas Compiláveis
💻 Exemplo 1 — Relacionamento 1:N (Autor / Livro) com Navegação Bidirecional
from typing import List, Optional
from sqlmodel import SQLModel, Field, Relationship, create_engine, Session
class Autor(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
nome: str
livros: List["Livro"] = Relationship(back_populates="autor")
class Livro(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
titulo: str
autor_id: int = Field(foreign_key="autor.id")
autor: Optional[Autor] = Relationship(back_populates="livros")
engine = create_engine("sqlite:///:memory:")
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
machado = Autor(nome="Machado de Assis")
machado.livros = [Livro(titulo="Dom Casmurro"), Livro(titulo="Memórias Póstumas")]
session.add(machado)
session.commit()
session.refresh(machado)
print(f"{machado.nome} possui {len(machado.livros)} livro(s): {[l.titulo for l in machado.livros]}")
💻 Exemplo 2 — Many-to-Many com Tabela de Ligação e Join Explícito
from typing import List, Optional
from sqlmodel import SQLModel, Field, Relationship, Session, create_engine, select
class UsuarioPerfilLink(SQLModel, table=True):
usuario_id: Optional[int] = Field(default=None, foreign_key="usuario.id", primary_key=True)
perfil_id: Optional[int] = Field(default=None, foreign_key="perfil.id", primary_key=True)
class Perfil(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
nome: str
usuarios: List["Usuario"] = Relationship(back_populates="perfis", link_model=UsuarioPerfilLink)
class Usuario(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
nome: str
perfis: List[Perfil] = Relationship(back_populates="usuarios", link_model=UsuarioPerfilLink)
engine = create_engine("sqlite:///:memory:")
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
admin = Perfil(nome="ROLE_ADMIN")
editor = Perfil(nome="ROLE_EDITOR")
usuario = Usuario(nome="Carla Reis", perfis=[admin, editor])
session.add(usuario)
session.commit()
stmt = select(Usuario.nome, Perfil.nome).join(UsuarioPerfilLink).join(Perfil)
for nome_usuario, nome_perfil in session.exec(stmt):
print(f"{nome_usuario} -> {nome_perfil}")
💻 Exemplo 3 — Eager Loading Assíncrono com selectinload (evitando N+1)
import asyncio
from typing import List, Optional
from sqlalchemy.orm import selectinload
from sqlmodel import SQLModel, Field, Relationship, select
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
class Autor(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
nome: str
livros: List["Livro"] = Relationship(back_populates="autor")
class Livro(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
titulo: str
autor_id: int = Field(foreign_key="autor.id")
autor: Optional[Autor] = Relationship(back_populates="livros")
async def main() -> None:
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
async with AsyncSession(engine) as session:
session.add(Autor(nome="Clarice Lispector", livros=[Livro(titulo="A Hora da Estrela")]))
session.add(Autor(nome="Jorge Amado", livros=[Livro(titulo="Gabriela"), Livro(titulo="Capitães da Areia")]))
await session.commit()
# Uma consulta para autores + uma consulta IN para livros = 2 no total, não N+1.
stmt = select(Autor).options(selectinload(Autor.livros))
resultado = await session.exec(stmt)
for autor in resultado.all():
print(f"{autor.nome}: {[l.titulo for l in autor.livros]}")
asyncio.run(main())