Starlette自定义文件校验中间件导致FastAPI HTTP通信停滞问题咨询

问题根因 #

你遇到的停滞问题由两个核心原因导致：

1. FastAPI底层依赖Starlette的HTTP请求体是流式只读结构，你在中间件中调用await request.form()后已经完整消费了请求体数据流，后续路由处理函数再次尝试读取请求体（比如获取上传文件、表单参数）时，会一直等待空数据流，最终导致请求挂起。
2. 你贴出的示例代码存在缩进错误：校验通过后执行response = await call_next(request)的代码缩进层级错误，部分场景下没有正常返回响应，也会导致请求挂起。

修复方案 #

你需要在消费完表单数据后，将读取到的请求体重新注入请求对象，让后续流程可以正常读取，修复后的代码如下：

from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.datastructures import FormData

class MyMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, file_type: list):
        super().__init__(app)
        self.file_type = file_type
        # 预定义不支持媒体类型响应，可根据业务需求调整
        self._unsupported_media_type = Response(
            content="Unsupported file type",
            status_code=415
        )

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        scope = request.scope

        if scope["method"] not in ("POST", "PUT"):
            return await call_next(request)

        # 缓存原始receive方法，读完请求体后重新注入
        original_receive = request._receive
        body_cache = b""
        async def cached_receive():
            nonlocal body_cache
            event = await original_receive()
            if event["type"] == "http.request":
                body_cache += event["body"]
                if not event.get("more_body", False):
                    # 读取完成后替换receive方法，直接返回缓存的请求体
                    async def return_cached():
                        return {"type": "http.request", "body": body_cache, "more_body": False}
                    request._receive = return_cached
            return event
        request._receive = cached_receive

        # 读取表单不会影响后续流程
        form: FormData = await request.form()
        # 先判断是否存在文件字段，避免空表单报错
        file_list = [item for item in form.values() if hasattr(item, "content_type")]
        if not file_list:
            # 无上传文件的请求可根据业务逻辑处理，这里直接放行
            return await call_next(request)
        # 校验所有上传文件的类型，原逻辑只校验第一个可按需调整
        for file in file_list:
            if file.content_type not in self.file_type:
                return self._unsupported_media_type

        # 校验通过放行请求
        return await call_next(request)

补充说明 #

• 如果你仅需要校验文件类型，也可以直接在路由函数的参数中通过File(...)的content_type参数实现校验，不需要自定义中间件，实现成本更低。
• 上述方案会缓存完整请求体，大文件上传场景下会占用较多内存，这种情况建议改用路由层校验或者反向代理层前置校验。

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