FastAPI接收POST数据遇JSON解码错误，无输入控制权如何解决？

问题重现 #

FastAPI 示例代码 #

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class RequestData(BaseModel):
    content: str

@app.post("/correct_json")
async def correct_json(data: RequestData):
    return {"processed": data.content}

错误的curl请求 #

curl -X POST "http://localhost:8000/correct_json" -H "Content-Type: application/json" -d '{"content": "This is a test with invalid escape: \z"}'

错误响应内容 #

{
  "detail": [
    {
      "loc": [
        "body"
      ],
      "msg": "Invalid JSON: invalid escape sequence at line 1 column 45",
      "type": "value_error.jsondecode",
      "ctx": {
        "msg": "invalid escape sequence",
        "doc": "{\"content\": \"This is a test with invalid escape: \\z\"}",
        "pos": 44
      }
    }
  ]
}

解决办法 #

方案1：自定义请求解析器预处理修复无效JSON #

替换FastAPI默认的JSON解析逻辑，先修复无效转义字符再解析：

from fastapi import FastAPI, Depends, Request
from pydantic import BaseModel
import json
import re

app = FastAPI()

class RequestData(BaseModel):
    content: str

# 修复非标准转义字符（比如\z这类无效转义）
def fix_invalid_json(raw_json: str) -> str:
    # 将单个\后跟非有效转义字符的情况替换为\\，让转义字符本身被保留
    return re.sub(r'\\(?![btnfr"\\/u])', r'\\\\', raw_json)

async def custom_json_parser(request: Request):
    raw_body = await request.body()
    try:
        return json.loads(raw_body)
    except json.JSONDecodeError:
        fixed_body = fix_invalid_json(raw_body.decode('utf-8'))
        return json.loads(fixed_body)

@app.post("/correct_json")
async def correct_json(data: RequestData = Depends(custom_json_parser)):
    return {"processed": data.content}

方案2：全局捕获JSON解码异常统一处理 #

通过FastAPI的异常处理器，拦截422错误中的JSON解码问题，尝试修复后再处理：

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel
import json
import re

app = FastAPI()

class RequestData(BaseModel):
    content: str

@app.exception_handler(RequestValidationError)
async def handle_json_decode_error(request: Request, exc: RequestValidationError):
    # 判断是否为JSON解码错误
    if any(err["type"] == "value_error.jsondecode" for err in exc.errors()):
        raw_body = await request.body()
        try:
            # 修复无效转义后重新解析
            fixed_str = re.sub(r'\\(?![btnfr"\\/u])', r'\\\\', raw_body.decode('utf-8'))
            parsed_data = json.loads(fixed_str)
            data = RequestData(**parsed_data)
            return JSONResponse(content={"processed": data.content}, status_code=200)
        except:
            return JSONResponse(
                content={"detail": "无效JSON，无法修复"},
                status_code=422
            )
    # 其他验证错误按默认逻辑返回
    return JSONResponse(content={"detail": exc.errors()}, status_code=422)

@app.post("/correct_json")
async def correct_json(data: RequestData):
    return {"processed": data.content}

方案3：直接接收原始请求体自行处理 #

跳过FastAPI自动解析，直接获取原始请求体，自己处理解析和修复逻辑：

from fastapi import FastAPI, Request
import json
import re

app = FastAPI()

@app.post("/correct_json")
async def correct_json(request: Request):
    raw_str = (await request.body()).decode('utf-8')
    try:
        parsed_data = json.loads(raw_str)
    except json.JSONDecodeError:
        # 尝试修复无效转义
        fixed_str = re.sub(r'\\(?![btnfr"\\/u])', r'\\\\', raw_str)
        try:
            parsed_data = json.loads(fixed_str)
        except:
            return {"detail": "无效JSON，修复失败"}
    content = parsed_data.get("content", "")
    return {"processed": content}

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