FastAPI中间件处理Form/JSON数据报错无法正常工作求助

你的中间件存在几个核心问题，直接导致了运行错误，以下是具体问题和修复后的代码：

核心错误点 #

1. 逻辑顺序完全颠倒：先调用call_next(request)让路由处理请求，之后再读取请求体，此时请求体已经被消费，会触发"Request body has already been consumed"错误。
2. Content-Type获取错误：从response.headers读取的是响应的类型，而非请求的Content-Type，完全搞反了对象。
3. 返回值不符合要求：BaseHTTPMiddleware的dispatch方法必须返回Response对象，你直接返回字典会触发类型错误。
4. 请求体重复读取：路由处理时已经读取过请求体，中间件后续再读取会报错。

修复后的Middleware.py #

from json import JSONDecodeError
from fastapi import HTTPException, Request
import urllib.parse
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.responses import Response

class MyMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        # 从请求头获取Content-Type（正确的来源）
        content_type = request.headers.get('Content-Type', '')
        
        # 先处理请求，再交给路由处理
        if content_type == 'application/json':
            try:
                # 提前验证JSON格式，避免路由处理时出错
                await request.json()
            except JSONDecodeError:
                raise HTTPException(status_code=400, detail='Invalid JSON data')
        elif content_type == 'application/x-www-form-urlencoded':
            try:
                # 读取并解析form数据
                payload = await request.body()
                parsed_data = urllib.parse.parse_qs(payload.decode())
                # 转换为{key: value}格式的字典（默认parse_qs返回{key: [value]}）
                form_data = {k: v[0] for k, v in parsed_data.items()}
                # 将处理后的数据存入request.state，供后续路由使用
                request.state.form_data = form_data
            except Exception:
                raise HTTPException(status_code=400, detail='Invalid Form data')
        # multipart/form-data交给FastAPI自带的Form/File依赖处理，无需中间件解析
        elif content_type.startswith('multipart/form-data'):
            pass
        # 非必要不要强制限制Content-Type，否则会拦截无请求体的合法请求
        # else:
        #     raise HTTPException(status_code=400, detail='Invalid data')
        
        # 处理完请求后，交给路由生成响应并返回
        response = await call_next(request)
        return response

路由中使用处理后的数据示例（user_route.py） #

from fastapi import APIRouter, Request

user_route = APIRouter()

@user_route.post("/user")
async def create_user(request: Request):
    # 从request.state获取中间件处理后的form数据
    data = getattr(request.state, "form_data", await request.json())
    # 业务逻辑处理
    return {"message": "User created successfully", "data": data}

关键修复说明 #

1. 调整执行顺序：先处理请求体验证和解析，再调用call_next让路由处理请求，避免请求体重复读取。
2. 正确传递数据：将解析后的form数据存入request.state，而非直接返回，保证中间件返回合法的Response对象。
3. 简化multipart处理：multipart/form-data类型的请求交给FastAPI原生的Form和File依赖处理，无需中间件手动解析。
4. 放宽类型限制：移除了强制拦截非指定Content-Type的逻辑，避免影响无请求体的GET等合法请求。

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