如何在FastAPI中根据查询参数修改列表的响应输出？

要实现根据查询参数动态切换FastAPI列表视图的响应格式，核心是动态选择Pydantic序列化模型，同时兼容fastapi-pagination的分页结构。以下是具体实现方案：

1. 定义差异化的序列化模型 #

首先创建基础版和扩展版的参会者模型，通过继承复用公共字段，减少代码重复：

from fastapi_pagination import Page, Params, paginate
from pydantic import BaseModel, Union
from sqlalchemy.orm import Session, joinedload
from fastapi import Depends, APIRouter

api = APIRouter()

class EventSerializer(BaseModel):
    id: str
    # 补充Event模型的其他字段
    class Config:
        orm_mode = True

# 基础模型：event字段为UUID字符串
class BaseEventAttendeeSerializer(BaseModel):
    id: str
    # 补充参会者的其他公共字段
    class Config:
        orm_mode = True

class EventAttendeeSerializer(BaseEventAttendeeSerializer):
    event: str

# 扩展模型：event字段为嵌套的EventSerializer
class EventAttendeeExtendedSerializer(BaseEventAttendeeSerializer):
    event: EventSerializer

2. 动态处理视图逻辑 #

在视图函数中根据extend参数判断是否需要扩展字段，选择对应模型序列化分页结果：

@api.get("/", response_model=Union[Page[EventAttendeeSerializer], Page[EventAttendeeExtendedSerializer]])
async def get_list(db: Session, pagination: Params = Depends(), extend: str = None):
    # 初始化查询
    query = db.query(myDbModel)
    
    # 判断是否需要扩展event字段，提前加载关联对象避免N+1查询
    should_extend_event = extend and "event" in extend.split(",")
    if should_extend_event:
        query = query.options(joinedload(myDbModel.event))
    
    # 执行查询并分页
    objects = query.all()
    paginated_result = paginate(objects, pagination)
    
    # 根据选择的模型序列化条目，构造分页响应
    if should_extend_event:
        serialized_items = [EventAttendeeExtendedSerializer.from_orm(item) for item in paginated_result.items]
    else:
        serialized_items = [EventAttendeeSerializer.from_orm(item) for item in paginated_result.items]
    
    return Page(
        items=serialized_items,
        total=paginated_result.total,
        page=paginated_result.page,
        size=paginated_result.size
    )

简化方案（可选） #

如果不想定义多个模型，也可以用Union字段配合SQLAlchemy关联加载实现动态序列化：

class EventAttendeeSerializer(BaseModel):
    id: str
    event: Union[str, EventSerializer]
    # 其他公共字段
    class Config:
        orm_mode = True
        smart_union = True  # 启用智能Union类型推断

@api.get("/", response_model=Page[EventAttendeeSerializer])
async def get_list(db: Session, pagination: Params = Depends(), extend: str = None):
    query = db.query(myDbModel)
    if extend and "event" in extend.split(","):
        query = query.options(joinedload(myDbModel.event))
    
    objects = query.all()
    return paginate(objects, pagination)

这种方式依赖Pydantic自动识别字段类型，适合字段较少、逻辑简单的场景。

关键注意事项 #

• 处理extend参数为None的情况，避免调用split()引发报错。
• 使用joinedload提前加载关联对象，避免SQLAlchemy懒加载导致的N+1查询问题。
• 用from_orm()方法确保ORM对象正确转换为Pydantic模型。

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