FastAPI中如何通过Depends实现请求体验证后自动构建上下文类

动机 #

使用Pydantic便捷验证请求体中的JSON数据，将验证后的Model实例传入上下文类以构建本次请求的上下文。上下文类通过@property实现的raw_params保留原始参数并禁止修改；同时提供可修改的curr_params以方便上下文传递。

当前现状 #

我可以在请求处理函数中手动构建上下文类的实例，但这会成为实际的样板代码（若要在项目中保持该行为一致），违背FastAPI减少代码重复的理念。

期望 #

通过FastAPI官方提供的Depends机制自动构建上下文类的实例。

最简实现代码 #

请注意，本示例已简化以隐藏隐私相关业务细节。

import copy
from typing import Generic, TypeVar
from uuid import uuid4

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class User(BaseModel):
    full_name: str
    nickname: str
    age: int


class Item(BaseModel):
    name: str
    price: float
    alias: str = ""
    brand: str = ""


T = TypeVar("T")


class QueryContext(Generic[T]):
    """简化上下文传递"""

    def __init__(self, raw_params: T):
        self.request_UUID = uuid4()
        self._raw_params = raw_params
        self.curr_params = copy.deepcopy(raw_params)
        self.extra = {}

    @property
    def raw_params(self):
        return self._raw_params

    @raw_params.setter
    def raw_params(self, *args, **kwargs):
        raise AttributeError("无法修改原始参数！请使用`curr_params`")


@app.post("/update_user")
async def update_user(user: User):
    ctx = QueryContext(user)  # 冗余代码
    ctx.curr_params.full_name = "new full name"
    ctx.curr_params.age += 5
    print(
        f"request {ctx.request_UUID} done. raw user: {ctx.raw_params} curr params: {ctx.curr_params}"
    )
    return ctx.curr_params


@app.post("/update_item")
async def update_item(item: Item):
    ctx = QueryContext(item)  # 冗余代码
    ctx.curr_params.name = "[HOT] " + ctx.curr_params.name
    ctx.curr_params.price = -1
    ctx.extra["msg"] = "Has sold out!"
    print(
        f"request {ctx.request_UUID} done. raw item: {ctx.raw_params} curr params: {ctx.curr_params} extra info: {ctx.extra}"
    )
    return ctx.curr_params

通过构建通用依赖函数，结合FastAPI的Depends机制，可以自动完成请求体验证和上下文实例化，消除样板代码。

修改后完整代码 #

import copy
from typing import Generic, TypeVar, Type
from uuid import uuid4

from fastapi import FastAPI, Depends
from pydantic import BaseModel

app = FastAPI()


class User(BaseModel):
    full_name: str
    nickname: str
    age: int


class Item(BaseModel):
    name: str
    price: float
    alias: str = ""
    brand: str = ""


T = TypeVar("T", bound=BaseModel)


class QueryContext(Generic[T]):
    """简化上下文传递"""

    def __init__(self, raw_params: T):
        self.request_UUID = uuid4()
        self._raw_params = raw_params
        self.curr_params = copy.deepcopy(raw_params)
        self.extra = {}

    @property
    def raw_params(self):
        return self._raw_params

    @raw_params.setter
    def raw_params(self, *args, **kwargs):
        raise AttributeError("无法修改原始参数！请使用`curr_params`")


def get_context(model_type: Type[T]) -> QueryContext[T]:
    """通用依赖函数，自动构建QueryContext实例"""
    async def dependency(params: model_type):
        return QueryContext(params)
    return dependency


@app.post("/update_user")
async def update_user(ctx: QueryContext[User] = Depends(get_context(User))):
    ctx.curr_params.full_name = "new full name"
    ctx.curr_params.age += 5
    print(
        f"request {ctx.request_UUID} done. raw user: {ctx.raw_params} curr params: {ctx.curr_params}"
    )
    return ctx.curr_params


@app.post("/update_item")
async def update_item(ctx: QueryContext[Item] = Depends(get_context(Item))):
    ctx.curr_params.name = "[HOT] " + ctx.curr_params.name
    ctx.curr_params.price = -1
    ctx.extra["msg"] = "已售罄！"
    print(
        f"request {ctx.request_UUID} done. raw item: {ctx.raw_params} curr params: {ctx.curr_params} extra info: {ctx.extra}"
    )
    return ctx.curr_params

核心说明 #

• 通用依赖函数get_context：接收Pydantic模型类型作为参数，返回一个异步依赖函数。FastAPI会自动将验证后的请求体实例传入该依赖函数，进而创建对应的QueryContext对象。
• 路由注入：在路由函数参数中通过Depends(get_context(ModelType))的方式注入上下文，无需手动实例化，彻底消除重复代码。
• 类型安全：通过泛型T绑定BaseModel，保证上下文与模型类型匹配，保留完整的类型提示能力。

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