如何将Pydantic模型转换为aws_cdk.aws_apigateway.JsonSchema？

核心思路 #

AWS CDK的aws_apigateway.JsonSchema是对标准JSON Schema的Python封装，仅在字段命名上做了适配（比如规避Python关键字的字段用下划线后缀、蛇形命名替代驼峰）。解决方案分为两步：

1. 从Pydantic模型导出标准JSON Schema
2. 将标准Schema的字段映射为CDK JsonSchema所需的命名格式，再实例化对象

具体实现代码 #

1. 定义Pydantic模型并导出标准Schema

from pydantic import BaseModel
from aws_cdk import aws_apigateway as apigw

# 示例Pydantic模型
class User(BaseModel):
    id: int
    name: str
    email: str | None = None
    is_active: bool = True

# 导出标准JSON Schema（Pydantic v2用model_json_schema，v1用schema）
pydantic_schema = User.model_json_schema()

2. 编写转换函数

递归处理标准Schema的所有字段，完成命名映射：

def convert_pydantic_to_cdk_schema(pydantic_schema: dict) -> apigw.JsonSchema:
    # 标准JSON Schema关键字到CDK JsonSchema参数的映射
    key_mapping = {
        "type": "type_",
        "definitions": "definitions_",
        "additionalProperties": "additional_properties",
        "patternProperties": "pattern_properties",
        "minProperties": "min_properties",
        "maxProperties": "max_properties",
        "dependencies": "dependencies_",
        "minItems": "min_items",
        "maxItems": "max_items",
        "uniqueItems": "unique_items",
        "contains": "contains_",
        "minLength": "min_length",
        "maxLength": "max_length",
        "pattern": "pattern_",
        "minimum": "minimum_",
        "maximum": "maximum_",
        "exclusiveMinimum": "exclusive_minimum",
        "exclusiveMaximum": "exclusive_maximum",
        "multipleOf": "multiple_of",
        "format": "format_",
        "default": "default_",
        "$ref": "ref_"
    }

    def _recursive_convert(schema: dict) -> dict:
        converted = {}
        for key, value in schema.items():
            cdk_key = key_mapping.get(key, key)
            if isinstance(value, dict):
                converted[cdk_key] = _recursive_convert(value)
            elif isinstance(value, list):
                converted[cdk_key] = [
                    _recursive_convert(item) if isinstance(item, dict) else item
                    for item in value
                ]
            else:
                converted[cdk_key] = value
        return converted

    converted_dict = _recursive_convert(pydantic_schema)
    return apigw.JsonSchema(**converted_dict)

3. 在CDK中使用转换后的Schema

将转换后的Schema用于API网关的请求校验：

# 转换为CDK JsonSchema
cdk_user_schema = convert_pydantic_to_cdk_schema(pydantic_schema)

# 创建请求校验器
request_validator = apigw.RequestValidator(
    self, "UserRequestValidator",
    rest_api=your_rest_api,
    validate_request_body=True
)

# 创建API模型并绑定到POST方法
user_model = apigw.Model(
    self, "UserModel",
    rest_api=your_rest_api,
    schema=cdk_user_schema,
    content_type="application/json"
)

your_api_resource.add_method(
    "POST",
    apigw.HttpIntegration("https://your-backend.example.com"),
    request_models={"application/json": user_model},
    request_validator=request_validator
)

注意事项 #

• Pydantic版本差异：v2使用model_json_schema()导出Schema，v1使用schema()，需根据实际版本调整。
• 嵌套模型处理：转换函数支持递归处理嵌套模型、数组内的对象结构，适配Pydantic导出的复杂Schema。
• 特殊字段兼容：已覆盖常见的JSON Schema关键字映射，若有遗漏可自行补充到key_mapping字典中。

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