FastAPI 中读取任意有效的 JSON 请求体

问题描述

在 FastAPI 开发中，我们经常需要接收 JSON 格式的请求体。但有时需求不仅仅是处理特定结构的对象或数组，还需要支持任意有效的 JSON 数据，包括：

• 基本数据类型：数字、字符串、布尔值、null
• 复杂数据结构：对象、数组
• 混合类型结构

使用 Pydantic 模型虽然能处理结构化数据，但对于完全任意的 JSON 数据就显得局限。那么如何在不牺牲 FastAPI 验证功能的情况下，灵活处理任意 JSON 请求体呢？

最佳解决方案

方法一：使用 Any 类型与 Body 注解

这是官方推荐的方法，利用了 FastAPI 的内置功能：

from typing import Any
from fastapi import FastAPI, Body

app = FastAPI()

@app.post('/test')
async def process_json(payload: Any = Body(...)):
    return payload

关键说明

• Any 类型允许接收任何有效的 JSON 数据类型
• Body(...) 中的 ... 表示该参数是必需的
• FastAPI 会自动验证输入是否为有效 JSON，无效时会返回标准验证错误
• 这种方法支持所有 JSON 数据类型：null、true、false、字符串、数字、对象和数组

方法二：使用 Annotated 注解（Python 3.9+）

对于更新版本的 Python，可以使用更现代的语法：

from typing import Any
from fastapi import FastAPI, Body
from typing_extensions import Annotated  # 对于 Python < 3.9

app = FastAPI()

@app.post("/test")
async def get_body(payload: Annotated[Any, Body()]):
    return payload

其他可行方案

方案三：使用 Request 对象直接获取

from fastapi import Request, FastAPI

app = FastAPI()

@app.post("/raw-json")
async def get_body(request: Request):
    return await request.json()

注意事项

这种方法虽然灵活，但：

• 需要手动处理异常
• 失去了自动的请求验证和文档生成功能
• 建议仅在需要完全控制请求处理时使用

方案四：使用 Dict 类型接收对象

from typing import Dict, Any
from fastapi import FastAPI

app = FastAPI()

@app.post("/dict")
async def post_dict(data: Dict[str, Any]):
    return data

局限性

这种方法只能接收 JSON 对象，无法处理简单值（如数字、字符串）或数组

验证测试示例

确保你的解决方案正确处理各种 JSON 数据类型：

import pytest
from fastapi.testclient import TestClient
from your_app import app  # 导入你的 FastAPI 应用

client = TestClient(app)

test_data = [
    ('{"key": 1}', {"key": 1}, "简单对象"),
    ('{"key": null}', {"key": None}, "包含 null 的对象"),
    ('[1, 2, 3]', [1, 2, 3], "数字数组"),
    ('"hello"', "hello", "纯字符串"),
    ('42', 42, "纯数字"),
    ('true', True, "布尔值 true"),
    ('false', False, "布尔值 false"),
    ('null', None, "纯 null"),
    ('[]', [], "空数组"),
    ('{}', {}, "空对象")
]

@pytest.mark.parametrize("payload,expected,test_name", test_data)
def test_json_variations(payload, expected, test_name):
    response = client.post("/test", content=payload)
    assert response.status_code == 200
    assert response.json() == expected

实际应用场景

场景一：通用 Webhook 处理器

from typing import Any
from fastapi import FastAPI, Body, HTTPException

app = FastAPI()

@app.post("/webhook/{service_name}")
async def webhook_handler(service_name: str, payload: Any = Body(...)):
    try:
        # 根据不同的服务处理不同的 JSON 结构
        if service_name == "github":
            return process_github_webhook(payload)
        elif service_name == "stripe":
            return process_stripe_webhook(payload)
        else:
            return {"status": "received", "data": payload}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

场景二：JSON 数据存储服务

from typing import Any
from fastapi import FastAPI, Body
import json
from datetime import datetime

app = FastAPI()

@app.post("/store-json")
async def store_json_document(
    document: Any = Body(...),
    collection: str = "default"
):
    # 生成文档ID和时间戳
    doc_id = generate_id()
    timestamp = datetime.utcnow()
    
    # 存储任意JSON数据
    store_to_database({
        "id": doc_id,
        "collection": collection,
        "data": document,
        "created_at": timestamp
    })
    
    return {"id": doc_id, "status": "stored"}

常见问题解答

Q: 如何处理非 JSON 请求体？

A: FastAPI 会自动验证请求内容类型。如果收到非 JSON 数据，会返回 422 错误，包含详细的验证信息。

Q: 如何设置可选的 JSON 请求体？

A: 将 Body(...) 改为 Body(None)：

async def optional_json(payload: Any = Body(None)):
    if payload is None:
        return {"message": "No data received"}
    return payload

Q: 这种方法会影响 API 文档吗？

A: 不会。FastAPI 会自动生成正确的 OpenAPI 文档，将请求体显示为接受任意 JSON。

总结

在 FastAPI 中接收任意有效的 JSON 数据有多种方法：

1. 推荐方法：使用 Any 类型与 Body 注解（自动验证 + 文档生成）
2. 备用方案：使用 Request 对象（完全控制但需手动处理）
3. 限制方案：使用 Dict[str, Any]（仅限对象类型）

根据你的具体需求选择合适的方法。对于大多数情况，方法一是最佳选择，因为它保持了 FastAPI 的所有优势：自动验证、序列化和文档生成，同时提供最大的灵活性。