FastAPI 0.79.1：中间件获取请求JSON体报错求助

问题背景 #

使用FastAPI 0.79.1版本，需通过中间件获取请求的JSON/字典类型数据，提取username和password字段。调用request.body()能获取字节类型数据，但测试代码执行后POST请求触发JSONDecodeError导致中断。

原测试代码 #

from fastapi import Request, FastAPI
import requests

app = FastAPI()

async def set_body(request: Request):
    receive_ = await request._receive()

    async def receive() -> Message:
        return receive_

    request._receive = receive

@app.middleware("http")
async def demo(request: Request, call_next):
    await set_body(request)
    jsob_body = await request.json()
    print(jsob_body)
    response = await call_next(request)
    return response

@app.post("/token")
async def login():
    return {"access_token": "test_token", "token_type": "bearer"}

# 请求测试代码
url = "http://127.0.0.1:8000/token"

payload={'username': 'test',
'password': 'test_pw'}
files=[]
headers = {}

response = requests.request("POST", url, headers=headers, data=payload, files=files)

报错信息 #

Traceback (most recent call last):
  File "D:\devtools\anaconda3\envs\fastapi\lib\site-packages\uvicorn\protocols\http\httptools_impl.py", line 372, in run_asgi
    result = await app(self.scope, self.receive, self.send)
  File "D:\devtools\anaconda3\envs\fastapi\lib\site-packages\uvicorn\middleware\proxy_headers.py", line 75, in __call__
    return await self.app(scope, receive, send)
  File "D:\devtools\anaconda3\envs\fastapi\lib\site-packages\fastapi\applications.py", line 269, in __call__
    await super().__call__(scope, receive, send)
  File "D:\devtools\anaconda3\envs\fastapi\lib\site-packages\starlette\applications.py", line 124, in __call__
    await self.middleware_stack(scope, receive, send)
  File "D:\devtools\anaconda3\envs\fastapi\lib\site-packages\starlette\middleware\errors.py", line 184, in __call__
    raise exc
  File "D:\devtools\anaconda3\envs\fastapi\lib\site-packages\starlette\middleware\errors.py", line 162, in __call__
    await self.app(scope, receive, _send)
  File "D:\devtools\anaconda3\envs\fastapi\lib\site-packages\starlette\middleware\base.py", line 68, in __call__
    response = await self.dispatch_func(request, call_next)
  File "D:\codes\ocr_server\.\main.py", line 43, in middle
    jsob_body = await request.json()
  File "D:\devtools\anaconda3\envs\fastapi\lib\site-packages\starlette\requests.py", line 242, in json
    self._json = json.loads(body)
  File "D:\devtools\anaconda3\envs\fastapi\lib\json\__init__.py", line 357, in loads
    return _default_decoder.decode(s)
  File "D:\devtools\anaconda3\envs\fastapi\lib\json\decoder.py", line 337, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "D:\devtools\anaconda3\envs\fastapi\lib\json\decoder.py", line 355, in raw_decode
    raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

问题原因 #

1. 请求格式不匹配：测试代码用requests的data参数提交数据，会以application/x-www-form-urlencoded格式发送，而非FastAPIrequest.json()能解析的application/json格式，导致解析失败。
2. 中间件操作内部API风险：直接修改request._receive属于操作Starlette内部未公开API，可能导致请求体读取异常或被提前耗尽。

解决方案 #

方案1：修正请求格式，发送JSON数据 #

将测试代码中的data参数替换为json，让请求以JSON格式发送：

# 修改后的请求测试代码
response = requests.request("POST", url, headers=headers, json=payload)

方案2：中间件兼容多请求格式 #

若需同时支持JSON和表单格式请求，可在中间件中判断请求类型后选择解析方式，同时缓存请求体供后续路由使用：

from fastapi import Request, FastAPI
import json

app = FastAPI()

async def get_parsed_body(request: Request):
    # 缓存原始请求体，避免重复读取
    if not hasattr(request, "_cached_body"):
        request._cached_body = await request.body()
    
    content_type = request.headers.get("Content-Type", "")
    parsed_body = {}
    
    if "application/json" in content_type:
        try:
            parsed_body = json.loads(request._cached_body)
        except json.JSONDecodeError:
            pass
    elif "application/x-www-form-urlencoded" in content_type:
        form_data = await request.form()
        parsed_body = dict(form_data)
    
    return parsed_body

@app.middleware("http")
async def demo(request: Request, call_next):
    req_body = await get_parsed_body(request)
    # 提取目标字段
    username = req_body.get("username")
    password = req_body.get("password")
    print(f"username: {username}, password: {password}")
    
    # 重新设置请求体，确保后续路由能正常读取
    async def receive():
        return {"type": "http.request", "body": request._cached_body}
    
    request._receive = receive
    
    response = await call_next(request)
    return response

@app.post("/token")
async def login():
    return {"access_token": "test_token", "token_type": "bearer"}

方案3：使用Starlette安全缓存请求体 #

基于Starlette的内置机制缓存请求体，避免直接操作内部API：

from fastapi import Request, FastAPI
from starlette.requests import Request as StarletteRequest
import json

app = FastAPI()

async def cache_body(request: StarletteRequest):
    if not hasattr(request, "_cached_body"):
        request._cached_body = await request.body()
    return request._cached_body

@app.middleware("http")
async def demo(request: Request, call_next):
    body = await cache_body(request)
    content_type = request.headers.get("Content-Type", "")
    req_body = {}
    
    if "application/json" in content_type:
        try:
            req_body = json.loads(body)
        except json.JSONDecodeError:
            pass
    elif "application/x-www-form-urlencoded" in content_type:
        form_data = await request.form()
        req_body = dict(form_data)
    
    username = req_body.get("username")
    password = req_body.get("password")
    print(f"username: {username}, password: {password}")
    
    # 重置请求体读取逻辑
    async def receive():
        return {"type": "http.request", "body": body}
    
    request._receive = receive
    
    response = await call_next(request)
    return response

@app.post("/token")
async def login():
    return {"access_token": "test_token", "token_type": "bearer"}

内容的提问来源于stack exchange，提问作者姚佳优