FastAPI+Pydantic v2+SQLAlchemy如何正确返回嵌套关联数据？

问题背景 #

在Pydantic v1中，通过orm_mode=True可以自动触发SQLAlchemy懒加载的关联属性（如Company关联的Programs）并序列化返回。但升级到Pydantic v2后，orm_mode被弃用，改用from_attributes配置，此时默认不会触发懒加载，导致关联字段被跳过，无法返回嵌套数据。当前通过手动调用model_validate触发加载的方式可行，但存在更优方案。

优化方案 #

方案1：查询时主动预加载关联数据（推荐） #

SQLAlchemy提供了selectinload、joinedload等预加载方法，在查询阶段直接加载关联数据，从根源避免懒加载问题，同时还能解决N+1查询的性能隐患。

修改接口查询代码：

from sqlalchemy.orm import selectinload

@router.get("/company/{company_id}")
def get_company(company_id: int, db: Session = Depends(get_db), response_model=CompanyAndProgramsSchema):
    # 使用selectinload预加载programs关联
    db_company = db.query(Company)\
        .options(selectinload(Company.programs))\
        .filter(Company.id == company_id)\
        .first()
    return db_company

• selectinload适合一对多关联场景，会生成独立的IN查询加载关联数据，性能优于懒加载；
• 预加载完成后，Pydantic的from_attributes可以直接识别已加载的关联属性，无需额外处理即可序列化返回嵌套数据。

方案2：在Pydantic模型中显式触发关联加载 #

如果无法修改查询语句，可以在Pydantic模型中通过自定义方法或验证器触发SQLAlchemy的懒加载：

方式A：重写from_orm方法

class CompanyAndProgramsSchema(BaseModel):
    name: str = Field(..., min_length=1, max_length=50)
    company_type_id: int
    programs: list[ProgramSchema]

    model_config = {
        "from_attributes": True
    }

    @classmethod
    def from_orm(cls, obj):
        # 显式访问关联属性触发懒加载
        if hasattr(obj, "programs"):
            # 转换为列表触发加载
            _ = list(obj.programs)
        return super().from_orm(obj)

方式B：使用字段验证器

from pydantic import BeforeValidator
from sqlalchemy.orm import InstrumentedList

def load_lazy_relationship(value):
    # 检测SQLAlchemy懒加载集合并触发加载
    if isinstance(value, InstrumentedList):
        list(value)
    return value

class CompanyAndProgramsSchema(BaseModel):
    name: str = Field(..., min_length=1, max_length=50)
    company_type_id: int
    programs: list[ProgramSchema] = Field(validators=[BeforeValidator(load_lazy_relationship)])

    model_config = {
        "from_attributes": True
    }

• 此方式无需修改查询逻辑，但要注意懒加载可能带来的N+1查询性能问题，适合小规模数据场景。

方案3：创建全局基础ORM模型 #

如果多个模型都需要处理关联加载，可以创建一个基础模型，统一处理懒加载触发逻辑，后续所有模型继承该基础模型即可：

from sqlalchemy.orm import InstrumentedList

class BaseOrmModel(BaseModel):
    model_config = {
        "from_attributes": True
    }

    @classmethod
    def from_orm(cls, obj):
        # 遍历模型字段，自动触发所有关联属性的懒加载
        for field_name in cls.model_fields:
            if hasattr(obj, field_name):
                attr = getattr(obj, field_name)
                if isinstance(attr, InstrumentedList):
                    list(attr)
        return super().from_orm(obj)

# 继承基础模型，自动处理关联加载
class CompanyAndProgramsSchema(BaseOrmModel):
    name: str = Field(..., min_length=1, max_length=50)
    company_type_id: int
    programs: list[ProgramSchema]

• 此方式一劳永逸，但同样要注意N+1查询的性能风险，建议仅在所有关联数据都需要返回的场景使用。

总结 #

• 优先选择方案1，主动预加载关联数据既能保证嵌套数据正确返回，又能优化查询性能；
• 方案2和3适合无法修改查询语句的场景，但需评估性能影响，必要时结合批量查询优化。

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