如何在SQLModel与FastAPI中基于URL参数条件加载关联实体？

核心思路 #

先明确两个关键点：

1. 禁用懒加载：SQLModel中Relationship默认lazy='raise'，正好符合需求——未显式加载关联时访问会报错，避免意外触发数据库查询。
2. 按需加载：根据URL参数include_heroes决定是否在查询阶段显式加载关联，再对应返回合适的模型或填充默认值。

方案1：拆分读取模型（最直观） #

定义两个读取模型，一个不带关联，一个带关联，根据参数动态选择返回：

from sqlmodel import SQLModel, Relationship, Session, select, selectinload
from fastapi import FastAPI, Query, Depends, HTTPException

# 数据库模型
class HeroBase(SQLModel):
    name: str
    team_id: int | None = None

class Hero(HeroBase, table=True):
    id: int | None = None
    team: "Team" = Relationship(back_populates="heroes")

class TeamBase(SQLModel):
    name: str

class Team(TeamBase, table=True):
    id: int | None = None
    # 保持默认lazy='raise'，禁用懒加载
    heroes: list[Hero] = Relationship(back_populates="team")

# 读取模型
class TeamRead(TeamBase):
    id: int

class TeamReadWithHeroes(TeamRead):
    heroes: list[HeroBase] = []  # 用HeroBase避免循环引用，默认空列表

app = FastAPI()

# 假设这里有获取Session的依赖
def get_session():
    with Session(engine) as session:
        yield session

@app.get("/teams/{team_id}")
def get_team(
    team_id: int,
    include_heroes: bool = Query(False),
    session: Session = Depends(get_session)
):
    # 基础查询
    query = select(Team).where(Team.id == team_id)
    
    if include_heroes:
        # 显式加载关联
        query = query.options(selectinload(Team.heroes))
        team = session.exec(query).first()
        return TeamReadWithHeroes.model_validate(team)
    
    team = session.exec(query).first()
    if not team:
        raise HTTPException(status_code=404, detail="Team not found")
    return TeamRead.model_validate(team)

方案2：单模型 + 动态查询（更简洁） #

复用同一个读取模型，通过查询阶段的条件加载控制返回内容：

# 复用上面的数据库模型，定义一个通用读取模型
class TeamRead(SQLModel):
    id: int
    name: str
    heroes: list[HeroBase] = []  # 默认空列表

@app.get("/teams/{team_id}")
def get_team(
    team_id: int,
    include_heroes: bool = Query(False),
    session: Session = Depends(get_session)
):
    query = select(Team).where(Team.id == team_id)
    
    if include_heroes:
        query = query.options(selectinload(Team.heroes))
    
    team = session.exec(query).first()
    if not team:
        raise HTTPException(status_code=404, detail="Team not found")
    
    # 未加载heroes时，模型会使用默认空列表
    return TeamRead.model_validate(team)

关键说明 #

• selectinload用于批量加载关联数据，避免N+1查询问题；如果是一对一关联，也可以用joinedload做关联查询。
• 因为Relationship默认lazy='raise'，如果没显式加载就访问team.heroes会报错，但我们的读取模型里heroes有默认值，model_validate会直接用默认值填充，不会触发关联访问。

方案3：利用Pydantic上下文（进阶） #

如果需要更灵活的逻辑，可以借助Pydantic的model_validate上下文参数动态处理：

class TeamRead(SQLModel):
    id: int
    name: str
    heroes: list[HeroBase] = []

    @classmethod
    def model_validate(cls, obj: Team, *, context: dict | None = None):
        if context and context.get("include_heroes"):
            session = context["session"]
            # 手动加载关联数据
            obj.heroes = session.exec(select(Hero).where(Hero.team_id == obj.id)).all()
        return super().model_validate(obj)

@app.get("/teams/{team_id}")
def get_team(
    team_id: int,
    include_heroes: bool = Query(False),
    session: Session = Depends(get_session)
):
    team = session.get(Team, team_id)
    if not team:
        raise HTTPException(status_code=404, detail="Team not found")
    
    return TeamRead.model_validate(
        team,
        context={"include_heroes": include_heroes, "session": session}
    )

这个方案适合需要在模型层面统一处理关联加载逻辑的场景，但相比前两种，代码稍复杂。

内容的提问来源于stack exchange，提问作者samidarko