如何修改SQLAlchemy查询实现FastAPI嵌套Pydantic序列化？

问题背景 #

使用SQLAlchemy 2.x和FastAPI开发时，无法让Pydantic正确序列化嵌套的用户角色结构。当前CRUD查询返回单个元组，无法识别嵌套关系，期望得到包含user_role嵌套对象的JSON响应。尝试过配置relationship()但未生效。

期望响应结构 #

{
  "id": 0,
  "username": "string",
  "avatar_url": "string",
  "create_at": "2023-03-30T10:56:03.625Z",
  "is_active": true,
  "user_role": {
    "name": "string",
    "color": "string"
  }
}

现有代码问题分析 #

1. CRUD查询未利用ORM关系：手动指定查询字段并返回元组，导致Pydantic无法识别嵌套的角色结构，同时join顺序错误（应该是从Users关联UsersRole，而非反向）。
2. Schema字段映射不匹配：期望响应的键是user_role，但Schema中定义的是role，与模型属性名一致但不符合输出要求。
3. 未预加载关联数据：即使配置了relationship()，如果查询时不预加载关联对象，会导致延迟加载（异步场景下更易出问题），无法正确序列化。

修复方案 #

1. 修改CRUD查询，利用ORM预加载关联数据 #

直接查询Users对象，通过joinedload预加载关联的role数据，确保返回完整的ORM对象，而非零散字段。

2. 调整Schema字段映射 #

修改ResponseCurrentUser，将role字段重命名为user_role，通过alias映射模型中的role属性。

3. 确保Relationship配置正确 #

模型中的relationship()配置本身没问题，但需配合预加载使用才能在查询时一次性获取关联数据。

完整修改代码 #

crud.py #

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import joinedload
from models import Users

async def get_current_user(user_id: int, session: AsyncSession):
    result = await session.execute(
        select(Users)
        .options(joinedload(Users.role))  # 预加载关联的角色数据
        .where(Users.id == user_id)
    )
    return result.scalar_one_or_none()  # 返回单个Users对象，或None

schemas.py #

from pydantic import BaseModel, Field
from datetime import datetime

class ResponseRoleUser(BaseModel):
    name: str
    color: str

class ResponseCurrentUser(BaseModel):
    id: int
    username: str
    avatar_url: str | None
    create_at: datetime
    is_active: bool
    user_role: ResponseRoleUser = Field(alias="role")  # 映射模型中的role属性为user_role

    class Config:
        orm_mode = True
        allow_population_by_field_name = True  # 允许通过字段名或别名赋值

endpoint.py（无需大改，确保返回ORM对象） #

from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Annotated
from crud import get_current_user
from schemas import ResponseCurrentUser
from database import get_async_session  # 假设你有这个依赖

router = APIRouter()

@router.get("/me", response_model=ResponseCurrentUser)
async def current_data_user(
    user_id: int,  # 假设通过认证获取user_id
    session: Annotated[AsyncSession, Depends(get_async_session)]
):
    user = await get_current_user(user_id, session)
    return user

验证说明 #

• 修改后的查询返回完整的Users对象，其中role属性已预加载UsersRole实例。
• Pydantic通过orm_mode自动将ORM对象转换为Schema结构，Field(alias="role")将模型的role属性映射为响应中的user_role键。
• 最终响应会完全符合期望的JSON嵌套结构。

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