如何在FastAPI中基于请求查询参数控制Pydantic Schema返回字段？

FastAPI本身没有原生支持通过response_fields查询参数动态指定Pydantic Schema返回字段的功能，但可以通过以下几种方式实现需求：

方案一：依赖项+手动筛选字段 #

通过自定义依赖项获取前端传入的字段列表，在路由中使用Pydantic模型的model_dump方法，通过include参数筛选返回字段，实现简单直接。

示例代码 #

from fastapi import FastAPI, Depends, Query, HTTPException
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class TestSchema(BaseModel):
    id: int
    name: str
    age: int

# 定义依赖项，获取并验证response_fields参数
def get_response_fields(
    response_fields: Optional[str] = Query(None, description="指定要返回的字段，多个用逗号分隔")
) -> Optional[List[str]]:
    if not response_fields:
        return None
    
    requested_fields = response_fields.split(",")
    valid_fields = set(TestSchema.model_fields.keys())
    invalid_fields = [f for f in requested_fields if f not in valid_fields]
    
    if invalid_fields:
        raise HTTPException(
            status_code=400,
            detail=f"无效字段：{', '.join(invalid_fields)}，可选字段：{', '.join(valid_fields)}"
        )
    return requested_fields

@app.get("/test")
def get_test(response_fields: Optional[List[str]] = Depends(get_response_fields)):
    # 模拟从数据库或其他数据源获取数据
    raw_data = TestSchema(id=1, name="张三", age=25)
    
    if response_fields:
        # 仅返回指定字段
        return raw_data.model_dump(include=set(response_fields))
    # 默认返回所有字段
    return raw_data

方案二：动态创建Pydantic模型 #

如果需要保持响应的模型验证（比如确保数据类型正确），可以通过create_model动态生成仅包含指定字段的Pydantic模型。

示例代码 #

from pydantic import create_model

@app.get("/test-dynamic")
def get_test_dynamic(response_fields: Optional[List[str]] = Depends(get_response_fields)):
    raw_data = TestSchema(id=1, name="张三", age=25)
    
    if response_fields:
        # 动态生成包含指定字段的模型
        DynamicResponseModel = create_model(
            "DynamicResponseModel",
            **{field: (TestSchema.model_fields[field].annotation, ...) for field in response_fields}
        )
        return DynamicResponseModel(**raw_data.model_dump(include=set(response_fields)))
    return raw_data

方案三：数据库层提前筛选（性能优化） #

如果数据来自数据库（如SQLAlchemy），可以直接在查询阶段只获取需要的字段，减少数据传输量，提升接口性能。

示例代码 #

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

# 初始化SQLAlchemy（示例配置）
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 定义数据库模型
class TestModel(Base):
    __tablename__ = "tests"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    age = Column(Integer)

Base.metadata.create_all(bind=engine)

# 依赖项获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/test-db")
def get_test_db(
    response_fields: Optional[List[str]] = Depends(get_response_fields),
    db: Session = Depends(get_db)
):
    if response_fields:
        # 仅查询指定字段
        query_fields = [getattr(TestModel, field) for field in response_fields]
        db_result = db.query(*query_fields).filter(TestModel.id == 1).first()
        return {field: getattr(db_result, field) for field in response_fields}
    
    # 默认查询所有字段并转为Pydantic模型
    db_result = db.query(TestModel).filter(TestModel.id == 1).first()
    return TestSchema.from_orm(db_result)

关键注意事项 #

• 字段校验：必须在依赖项中验证前端传入的字段是否存在于目标Pydantic模型中，避免无效字段导致的错误。
• 多字段支持：通过逗号分隔多个字段（如response_fields=id,name），满足前端同时请求多个字段的需求。

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