FastAPI自定义Schema未校验数据：非Pydantic模型下如何修复

我在自定义OpenAPI Schema中将price字段定义为number类型，但传入字符串值时，Schema既不执行校验也不抛出错误。如何在不使用Pydantic模型的前提下修改该行为？

原代码 #

def magic_data_reader(raw_body: bytes):
    raw_body = dict(eval(raw_body))
    return {
        "size": len(raw_body),
        "content": {
            "name": raw_body['name'],
            "price": raw_body['price'],
            "description": raw_body['description'],
        },
    }


@router.post(
    "/items/",
    openapi_extra={
        "requestBody": {
            "content": {
                "application/json": {
                    "schema": {
                        "required": ["name", "price"],
                        "type": "object",
                        "properties": {
                            "name": {"type": "string"},
                            "price": {"type": "number"},
                            "description": {"type": "string"},
                        },
                    }
                }
            },
            "required": True,
        },
    },
)
async def create_item(request: Request):
    raw_body = await request.body()
    data = magic_data_reader(raw_body)
    return data

测试请求 #

{
  "name": "Jack",
  "price": "dddd",
  "description": "Something"
}

当前响应 #

{
  "size": 3,
  "content": {
    "name": "Jack",
    "price": "dddd",
    "description": "Something"
  }
}

原因说明 #

FastAPI的openapi_extra仅用于生成OpenAPI文档，不会自动对请求数据执行校验逻辑。直接通过request.body()获取的原始数据需要手动处理校验。

方法一：手动添加类型校验与转换 #

在magic_data_reader函数中对字段逐一做类型检查，尝试转换price为数字，失败则抛出HTTP异常：

from fastapi import HTTPException
import json

def magic_data_reader(raw_body: bytes):
    # 替换eval为安全的json解析
    try:
        raw_body = json.loads(raw_body.decode("utf-8"))
    except json.JSONDecodeError:
        raise HTTPException(status_code=400, detail="无效的JSON格式")
    
    # 校验name字段类型
    if not isinstance(raw_body.get("name"), str):
        raise HTTPException(status_code=400, detail="name必须为字符串类型")
    
    # 校验并转换price字段
    price = raw_body.get("price")
    try:
        price = float(price)
    except (ValueError, TypeError):
        raise HTTPException(status_code=400, detail="price必须为有效的数字")
    
    # 校验description字段（可选）
    description = raw_body.get("description")
    if description is not None and not isinstance(description, str):
        raise HTTPException(status_code=400, detail="description必须为字符串类型")
    
    return {
        "size": len(raw_body),
        "content": {
            "name": raw_body['name'],
            "price": price,
            "description": description,
        },
    }

方法二：使用jsonschema库进行Schema校验 #

借助专业JSON Schema校验库，复用已定义的OpenAPI Schema完成批量校验：

1. 安装依赖：

pip install jsonschema

2. 修改代码：

from jsonschema import validate, ValidationError
from fastapi import HTTPException
import json

# 复用自定义的Schema
item_schema = {
    "required": ["name", "price"],
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "price": {"type": "number"},
        "description": {"type": "string"},
    },
}

def magic_data_reader(raw_body: bytes):
    try:
        raw_body = json.loads(raw_body.decode("utf-8"))
    except json.JSONDecodeError:
        raise HTTPException(status_code=400, detail="无效的JSON格式")
    
    # 执行Schema校验
    try:
        validate(instance=raw_body, schema=item_schema)
    except ValidationError as e:
        raise HTTPException(status_code=400, detail=f"校验失败: {e.message}")
    
    return {
        "size": len(raw_body),
        "content": {
            "name": raw_body['name'],
            "price": raw_body['price'],
            "description": raw_body['description'],
        },
    }

重要提示 #

原代码中使用eval解析JSON存在严重安全风险，已替换为json.loads()，避免代码注入攻击。

内容的提问来源于stack exchange，提问作者Роман Яровой