FastAPI同时处理camelCase与PascalCase请求体转snake_case问题

我尝试用中间件和路由处理器把camelCase转成PascalCase，但效果不好。以下是我的代码：

class CustomRouteHandler(APIRoute):

    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            route_path = request.url.path
            body = await request.body()
            logger.info({"path": request.url.path, "request": request._body.decode("utf-8")})
            if body:
                body = ujson.dumps(humps.pascalize(ujson.loads(body.decode("utf-8")))).encode("ascii")
            request._body = body
            try:
                return await original_route_handler(request)
            except ValidationError as e:
                logger.exception(e, exc_info=True)
                return UJSONResponse(status_code=200, content={"Success": False, "Message": e})
            
        return custom_route_handler

router = APIRouter(prefix="/payments", route_class=CustomRouteHandler)

运行后日志显示转换成功，但还是返回了ValidationError：

request body: {"test": 12345}
logger after pascalize: {"Test": 12345}
ERROR: 1 validation error for Request
body -> Test 
none is not an allowed value (type=type_error.none.not_allowed)

问题分析 #

1. 请求体读取失效：调用await request.body()后，请求的数据流已被耗尽，即便修改了私有属性request._body，FastAPI后续的解析逻辑也无法正确读取修改后的内容，导致模型校验时拿不到对应字段值。
2. 转换方向错误：需求是将PascalCase/camelCase转成snake_case，但当前代码是把输入转成PascalCase，完全偏离目标，即便解析正常也不符合预期。

解决方案 #

方案一：用Pydantic别名生成器（推荐）

直接通过Pydantic配置实现字段自动映射，无需修改请求体，简洁可靠：

from pydantic import BaseModel, ConfigDict
from humps import decamelize

# 定义转换函数，同时支持camelCase和PascalCase转snake_case
def to_snake_case(s: str) -> str:
    return decamelize(s)

# 基础请求模型，所有业务请求模型继承它
class BaseRequestModel(BaseModel):
    model_config = ConfigDict(
        alias_generator=to_snake_case,
        populate_by_name=True  # 允许同时通过别名和原字段名赋值
    )

# 示例业务请求模型
class PaymentRequest(BaseRequestModel):
    test: int  # 自动匹配请求中的test、Test等格式

在路由中直接使用该模型：

@router.post("/process")
async def process_payment(request: PaymentRequest):
    return {"received_test": request.test}

不管请求体是{"test":12345}、{"Test":12345}还是{"testCase":6789}，都能正确映射到模型的snake_case字段。

方案二：中间件转换请求体

如果必须通过中间件修改请求体，需要正确重置请求数据流，避免流耗尽问题：

from fastapi import Request, Response
from humps import decamelize
import ujson

async def case_conversion_middleware(request: Request, call_next) -> Response:
    # 仅处理JSON类型的POST/PUT/PATCH请求
    if request.method in ["POST", "PUT", "PATCH"] and request.headers.get("content-type") == "application/json":
        # 读取原始请求体并转换为snake_case
        raw_body = await request.json()
        snake_case_body = decamelize(raw_body)
        
        # 重新构造请求数据流
        async def modified_receive():
            return {"type": "http.request", "body": ujson.dumps(snake_case_body).encode("utf-8")}
        
        # 替换请求的receive方法
        request._receive = modified_receive
    
    response = await call_next(request)
    return response

将中间件添加到FastAPI应用：

from fastapi import FastAPI

app = FastAPI()
app.add_middleware(case_conversion_middleware)
app.include_router(router)

中间件会自动将符合条件的请求体转成snake_case，后续Pydantic模型可正常解析。

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