能否让FastAPI不反序列化Pydantic模型但仍用其生成OpenAPI规范？

问题背景 #

我用FastAPI写了个接口，处理大JSON请求时内存占用过高：

class Foo(BaseModel):
    very_large_something: list[SomeOtherModel]

app = FastAPI()

@app.post("/foos/")
async def create(foo: Foo):
    # 业务逻辑

20MB的请求反序列化后占了100MB内存，顶不住。想改用流式解析验证，但又不想丢了FastAPI自动生成OpenAPI文档的功能，理想写法是这样的：

@app.post("/foos/")
async def create(_unused: Annotated[Foo, DoNotDeserialize], request: Request):
    # 直接操作request，不用_unused

实现方法 #

1. 自定义跳过反序列化的依赖 #

先写个自定义依赖，让FastAPI只拿它来生成接口文档，不实际解析请求体：

from fastapi import Depends, Request, Body
from pydantic import BaseModel
from typing import Annotated, Any

class DoNotDeserialize:
    def __init__(self, model: type[BaseModel]):
        self.model = model

    def __call__(self, request: Request) -> Any:
        # 跳过反序列化，直接返回空值
        return None

# 封装成便捷的注解形式
def NoDeserialize(model: type[BaseModel]):
    return Annotated[Any, Depends(DoNotDeserialize(model)), Body(model=model)]

2. 修改接口定义 #

用自定义的NoDeserialize替代直接传Pydantic模型，既保留文档生成，又跳过反序列化：

class SomeOtherModel(BaseModel):
    # 子模型字段定义
    field1: str
    field2: int

class Foo(BaseModel):
    very_large_something: list[SomeOtherModel]

app = FastAPI()

@app.post("/foos/")
async def create(_unused: Annotated[Foo, NoDeserialize(Foo)], request: Request):
    # 这里写流式处理逻辑
    return {"status": "processed"}

3. 流式解析请求体 #

用流式JSON库逐块处理请求，避免一次性加载整个大对象：

import json_stream

async def process_large_request(request: Request):
    # 异步读取请求体
    body = await request.body()
    # 流式加载JSON数据
    json_data = json_stream.load(body)
    # 逐个处理列表中的子项
    for item in json_data["very_large_something"]:
        # 验证单个子模型
        validated_item = SomeOtherModel(**item)
        # 处理单个子项的业务逻辑
        print(validated_item.field1)

核心逻辑说明 #

• Body(model=model)会告诉FastAPI用指定的Pydantic模型生成OpenAPI的请求体Schema，保证接口文档正常生成。
• Depends(DoNotDeserialize)则跳过FastAPI默认的请求体反序列化流程，让我们可以直接操作原始请求流。
• 流式解析时需要自行完成模型验证，确保每条数据符合定义，这样既节省内存又不丢失校验逻辑。

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