基于含字段验证器的JSON Schema动态生成Pydantic模型及MongoDB存储方案

核心问题分析 #

原生JSON Schema不支持直接描述Pydantic的field_validator逻辑，因为验证器属于Python代码层面的行为。因此需要自定义扩展JSON Schema字段，用来标记验证器的标识或规则，再在生成模型时映射为对应的Python验证函数。

步骤1：定义带验证器元信息的JSON Schema结构 #

在JSON Schema中添加自定义字段（比如x-pydantic-validators）存储验证器配置，示例Schema可直接存入MongoDB：

{
  "title": "UserModel",
  "type": "object",
  "properties": {
    "name": {
      "type": "string",
      "x-pydantic-validators": [
        {"name": "name_must_contain_space"}
      ]
    },
    "id": {
      "type": "integer"
    }
  },
  "required": ["name", "id"]
}

步骤2：实现验证器映射库 #

提前定义所有可能用到的验证器函数，通过名称映射，从MongoDB取出Schema后可直接匹配对应验证器：

from pydantic import BaseModel, Field, field_validator, create_model
from typing import Dict, Callable

# 预定义验证器库，集中管理所有验证逻辑
VALIDATORS: Dict[str, Callable] = {
    "name_must_contain_space": field_validator('name')(
        lambda cls, v: v.title() if ' ' in v else ValueError('Name must contain a space')
    )
}

步骤3：编写动态模型生成函数 #

读取MongoDB中的Schema后，解析自定义验证器字段，将验证器绑定到动态模型：

def create_dynamic_model_from_schema(schema_data: Dict) -> BaseModel:
    # 提取模型基础信息
    model_name = schema_data.get("title", "DynamicModel")
    properties = schema_data.get("properties", {})
    required_fields = schema_data.get("required", [])

    # 构建字段字典
    fields = {}
    type_map = {"string": str, "integer": int, "number": float, "boolean": bool}
    for field_name, prop in properties.items():
        # 映射JSON Schema类型到Python类型
        field_type = type_map.get(prop["type"], str)
        
        # 构建Field参数
        field_kwargs = {"title": field_name}
        field_kwargs["default"] = ... if field_name in required_fields else None
        
        fields[field_name] = (field_type, Field(**field_kwargs))
    
    # 创建基础动态模型
    dynamic_model = create_model(model_name, **fields)
    
    # 绑定字段验证器到模型
    for field_name, prop in properties.items():
        validator_configs = prop.get("x-pydantic-validators", [])
        for config in validator_configs:
            validator_func = VALIDATORS.get(config["name"])
            if validator_func:
                setattr(dynamic_model, f"validate_{field_name}", validator_func)
    
    return dynamic_model

步骤4：MongoDB的Schema存储与读取 #

将上述JSON Schema作为文档存入MongoDB集合（如schema_collection），读取后传入生成函数即可：

from pymongo import MongoClient

# 连接MongoDB
client = MongoClient("mongodb://localhost:27017/")
db = client["your_database"]
schema_collection = db["schema_collection"]

# 读取目标Schema
schema_data = schema_collection.find_one({"title": "UserModel"})

# 生成动态模型
UserModel = create_dynamic_model_from_schema(schema_data)

# 测试模型逻辑
try:
    user = UserModel(name="JohnDoe", id=123)
except ValueError as e:
    print(e)  # 输出: Name must contain a space

valid_user = UserModel(name="John Doe", id=123)
print(valid_user.name)  # 输出: John Doe

最佳实践建议 #

• 验证器中心化管理：不要在MongoDB存储验证器代码，仅存标识，避免安全风险与序列化问题。
• Schema版本控制：在Schema文档中加入version字段，方便后续迭代更新模型。
• 扩展类型映射：可扩展type_map支持复杂类型（如List[str]），或用Pydantic的TypeAdapter解析更复杂的JSON Schema类型。
• 缓存动态模型：用functools.lru_cache缓存已生成的模型，避免重复创建提升性能。

替代方案 #

若需更灵活的验证逻辑存储，可考虑：

• 在可信环境下，用MongoDB存储验证器代码片段，读取后通过exec动态加载（生产环境不推荐）。
• 使用Pydantic的model_validator替代field_validator，在Schema中用自定义字段描述验证规则，再动态生成模型级验证器。

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