如何通用实现多Pydantic重叠字段模型的自动合并？

问题场景 #

现有以下Pydantic模型，需要构建一个包含AppleData1和AppleData2所有属性的Apple模型，且满足特定合并规则：

from pydantic import BaseModel

class Detail1(BaseModel):
    round: bool
    volume: float

class AppleData1(BaseModel):
    origin: str
    detail: Detail1

class Detail2(BaseModel):
    round: bool
    weight: float

class AppleData2(BaseModel):
    origin: str
    detail: Detail2

合并规则 #

• 同名属性处理：
  • 若类型相同，保留一个即可（如origin均为str，合并后仍为str）
  • 若类型不同，需合并属性为包含所有内部字段的子模型（如detail分别对应Detail1和Detail2，需合并为包含round、volume、weight的子模型）
• 子模型同名属性为同一物理量时允许覆盖（如Detail1.round和Detail2.round均为bool，合并后仍为bool）

通用实现方案 #

下面是无需硬编码子模型的通用合并函数，可处理任意层级的模型合并：

from pydantic import BaseModel
from typing import Dict, Type, Any

def merge_pydantic_models(*models: Type[BaseModel]) -> Type[BaseModel]:
    merged_fields: Dict[str, Any] = {}
    
    for model in models:
        for field_name, field in model.model_fields.items():
            field_type = field.annotation
            # 处理子模型合并
            if isinstance(field_type, type) and issubclass(field_type, BaseModel):
                if field_name in merged_fields:
                    existing_type = merged_fields[field_name].annotation
                    if isinstance(existing_type, type) and issubclass(existing_type, BaseModel):
                        # 递归合并子模型
                        merged_submodel = merge_pydantic_models(existing_type, field_type)
                        merged_fields[field_name] = field.model_copy(update={"annotation": merged_submodel})
                    else:
                        # 非模型类型冲突，保留当前模型的字段定义
                        merged_fields[field_name] = field
                else:
                    merged_fields[field_name] = field
            else:
                # 非模型字段，直接保留当前模型的定义（同类型保留一个，不同类型覆盖）
                merged_fields[field_name] = field
    
    # 动态生成合并后的模型类
    merged_model_name = f"Merged{''.join(m.__name__ for m in models)}"
    merged_model = type(merged_model_name, (BaseModel,), {"__annotations__": {k: v.annotation for k, v in merged_fields.items()}})
    # 复制原字段的所有配置（默认值、验证器等）
    for field_name, field in merged_fields.items():
        merged_model.model_fields[field_name] = field
    return merged_model

# 使用示例
Apple = merge_pydantic_models(AppleData1, AppleData2)

# 验证合并结果
print(Apple.schema_json(indent=2))

代码说明 #

1. 递归合并逻辑：函数遍历每个模型的字段，若字段类型是BaseModel子类，则递归调用自身合并子模型，支持多层级嵌套模型的合并。
2. 字段处理规则：非模型类型字段直接保留最后一个模型的定义，符合规则中同类型保留、不同类型覆盖的要求。
3. 动态模型创建：通过type()函数动态生成合并后的模型类，同时复制原字段的所有配置（默认值、验证器等），确保原模型的特性完整保留。

验证结果 #

运行代码后生成的Apple模型与预期的硬编码模型完全等价，其Schema输出如下：

{
  "title": "MergedAppleData1AppleData2",
  "type": "object",
  "properties": {
    "origin": {
      "title": "Origin",
      "type": "string"
    },
    "detail": {
      "$ref": "#/definitions/MergedDetail1Detail2"
    }
  },
  "required": [
    "origin",
    "detail"
  ],
  "definitions": {
    "MergedDetail1Detail2": {
      "title": "MergedDetail1Detail2",
      "type": "object",
      "properties": {
        "round": {
          "title": "Round",
          "type": "boolean"
        },
        "volume": {
          "title": "Volume",
          "type": "number"
        },
        "weight": {
          "title": "Weight",
          "type": "number"
        }
      },
      "required": [
        "round",
        "volume",
        "weight"
      ]
    }
  }
}

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