如何在Pydantic中限制model_dump的序列化深度？

要实现基于max_mapping_depth限制序列化深度的需求，可以通过自定义序列化逻辑+跟踪递归深度来实现，以下是两种可行的实现方式：

方式一：重写模型的model_dump方法，全局跟踪深度 #

这种方式适合在根模型中定义深度限制，递归序列化时自动判断深度：

from pydantic import BaseModel
from typing import ClassVar, Any

class Level3(BaseModel):
    deep_field: str
    id: str  # 假设每个子模型都有唯一标识字段

class Level2(BaseModel):
    mid_field: str
    level3: Level3
    id: str

class Level1(BaseModel):
    top_field: str
    level2: Level2
    id: str

class DepthLimitedModel(BaseModel):
    name: str
    level1: Level1
    max_mapping_depth: ClassVar[int] = 1

    def model_dump(self, *args, **kwargs) -> dict[str, Any]:
        # 自定义序列化逻辑，跟踪当前深度
        def serialize(obj: BaseModel, current_depth: int) -> dict:
            result = {}
            for field_name, field_value in obj.__dict__.items():
                if field_name.startswith('_'):  # 跳过Pydantic内部字段
                    continue
                if isinstance(field_value, BaseModel):
                    if current_depth < self.max_mapping_depth:
                        result[field_name] = serialize(field_value, current_depth + 1)
                    else:
                        # 超过深度时，仅保留对象的id（或其他标识）
                        result[field_name] = {"id": field_value.id, "__type__": type(field_value).__name__}
                else:
                    result[field_name] = field_value
            return result
        
        return serialize(self, current_depth=0)

# 测试
new_model = DepthLimitedModel(
    name="Test",
    level1=Level1(
        top_field="Top",
        id="level1_001",
        level2=Level2(
            mid_field="Mid",
            id="level2_001",
            level3=Level3(deep_field="Deep", id="level3_001")
        )
    )
)
dumped = new_model.model_dump()
print(dumped)

输出结果（max_mapping_depth=1时）：

{
    'name': 'Test',
    'level1': {
        'top_field': 'Top',
        'id': 'level1_001',
        'level2': {'id': 'level2_001', '__type__': 'Level2'}
    }
}

方式二：使用Pydantic的serializer参数，局部控制序列化 #

如果不想重写整个model_dump方法，可以给需要限制深度的字段指定自定义序列化器：

from pydantic import BaseModel, field_serializer
from typing import ClassVar, Any

class Level3(BaseModel):
    deep_field: str
    id: str

class Level2(BaseModel):
    mid_field: str
    level3: Level3
    id: str

class Level1(BaseModel):
    top_field: str
    level2: Level2
    id: str

class DepthLimitedModel(BaseModel):
    name: str
    level1: Level1
    max_mapping_depth: ClassVar[int] = 1

    @field_serializer('level1')
    def serialize_level1(self, value: Level1, info) -> dict:
        def serialize(obj: BaseModel, depth: int) -> dict:
            data = obj.model_dump(exclude_unset=True)
            for k, v in data.items():
                if isinstance(v, BaseModel):
                    if depth >= self.max_mapping_depth:
                        data[k] = {"id": v.id, "__type__": type(v).__name__}
                    else:
                        data[k] = serialize(v, depth + 1)
            return data
        return serialize(value, depth=0)

# 测试
new_model = DepthLimitedModel(
    name="Test",
    level1=Level1(
        top_field="Top",
        id="level1_001",
        level2=Level2(
            mid_field="Mid",
            id="level2_001",
            level3=Level3(deep_field="Deep", id="level3_001")
        )
    )
)
dumped = new_model.model_dump()
print(dumped)

关键说明 #

• 标识字段：示例中假设每个子模型都有id字段作为唯一标识，如果你的模型没有这个字段，可以替换为对象的内存ID（id(field_value)）或者其他自定义标识。
• 深度计算：示例中current_depth从0开始，根模型（DepthLimitedModel）为深度0，其直接子模型（Level1）为深度1，当current_depth等于max_mapping_depth时，停止递归序列化子模型。
• 灵活性：如果需要更通用的解决方案，可以封装一个通用的深度序列化函数，让所有模型都能复用，而不需要在每个根模型中重复编写逻辑。

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