FastAPI类视图访问Request请求体报Stream consumed错误解决方法

FastAPI 接收的请求体是单向只读的字节流，全程仅可被消费一次。
你在 CBV 类中声明了类级属性 file: Optional[UploadFile] = File(None)，fastapi-utils 的 CBV 实现会把类级声明的参数作为全局依赖注入，FastAPI 在执行路由函数前，会自动解析这个 File 类型参数——解析过程会完整读取请求体流，等路由内你再调用 await self.request.json() 尝试读取请求体时，流已经被消费完，就会抛出 RuntimeError: Stream consumed 错误。
另外你的测试 curl 命令也存在参数错误：-x 是指定代理的参数，发送 POST 请求应该用大写 -X；如果要传 JSON 格式请求体，需要额外指定 Content-Type: application/json 请求头；如果要传 File 文件，请求编码必须是 multipart/form-data，和 JSON 格式的请求体无法原生共存。

根据你的业务场景选对应方案即可：

• 方案1：不要手动调用 request.json() 解析请求体，所有请求参数（包括 JSON 字段、文件、表单参数）都通过 FastAPI 内置的 Body/File/Form 声明式写法定义，框架会自动完成解析，不会出现重复消费流的问题。
  示例修正代码：

  from typing import Optional
  from fastapi import FastAPI, File, Request, UploadFile, Body
  from fastapi_utils.inferring_router import InferringRouter
  from fastapi_utils.cbv import cbv
  
  
  app = FastAPI()
  router = InferringRouter()
  
  @cbv(router)
  class ExampleRouteClass:
      request: Request
      file: Optional[UploadFile] = File(None)
      # 声明JSON请求体字段
      some_data: Optional[str] = Body(None)
      
      @router.post("/example-route/")
      async def example_function(self):
          headers = self.request.headers
          # 直接通过self属性获取框架解析好的参数，不需要手动读request.json()
          return {"headers": headers, "some_data": self.some_data, "file_name": self.file.filename if self.file else None}
  
  app.include_router(router)
  

• 方案2：如果你确实需要手动操作 Request 对象读取原始请求体，就不要把 File/Form 这类会触发请求体解析的参数声明为类级属性，把这类参数下沉到具体需要用到的路由方法内作为参数，避免框架提前消费请求流。
• 方案3：如果必须在类级声明 File/Form 参数，同时还要读取原始请求体，可以添加全局中间件提前缓存请求体，让请求流支持重复读取。注意该方案会把完整请求体加载到内存中，上传大文件场景谨慎使用。
  中间件示例代码：

  from starlette.middleware.base import BaseHTTPMiddleware
  
  class CacheBodyMiddleware(BaseHTTPMiddleware):
      async def dispatch(self, request: Request, call_next):
          # 提前读取完整请求体缓存
          body = await request.body()
          # 重写request的receive方法，返回缓存的内容
          async def receive():
              return {"type": "http.request", "body": body}
          request._receive = receive
          # 继续执行后续逻辑
          return await call_next(request)
  
  # 注册中间件到app
  app.add_middleware(CacheBodyMiddleware)
  

  修正后的curl测试命令：

  curl -X POST 'http://127.0.0.1:端口号/example-route' \
  -H 'secret-key: supersecret' \
  -H 'Content-Type: application/json' \
  -d '{"some_data": "data"}'
  

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