Pydantic递归模型深度校验36计：从无限嵌套到亿级数据的优化法则

from pydantic import BaseModel
from typing import List, Optional


class TreeNode(BaseModel):
    name: str
    children: List['TreeNode'] = []  # 前向引用


# 创建7层深度树结构
root = TreeNode(name="root", children=[
    TreeNode(name="L1", children=[
        TreeNode(name="L2", children=[
            TreeNode(name="L3")
        ])
    ])
])

class GraphNode(BaseModel):
    id: str
    edges: List['GraphNode'] = []

    @validator('edges')
    def check_cycles(cls, v, values):
        visited = set()

        def traverse(node, path):
            if node.id in path:
                raise ValueError(f"环状路径检测: {'->'.join(path)}->{node.id}")
            if node.id not in visited:
                visited.add(node.id)
                for edge in node.edges:
                    traverse(edge, path + [node.id])

        traverse(values['self'], [])
        return v

from pydantic import Field


class User(BaseModel):
    id: int
    friends: List['User'] = Field(default_factory=list)
    manager: Optional['User'] = None

    @root_validator
    def validate_relationships(cls, values):
        def check_hierarchy(user: User, seen=None):
            seen = seen or set()
            if user.id in seen:
                raise ValueError("管理关系循环")
            seen.add(user.id)
            if user.manager:
                check_hierarchy(user.manager, seen)

        check_hierarchy(values['self'])
        return values

class LazyValidator(BaseModel):
    data: str
    _parsed: dict = None

    @validator('data', pre=True)
    def lazy_parse(cls, v):
        # 延迟解析直到首次访问
        instance = cls()
        instance._parsed = json.loads(v)
        return v

    @root_validator
    def validate_content(cls, values):
        if values['_parsed'] is None:
            values['_parsed'] = json.loads(values['data'])
        # 执行深度校验逻辑
        validate_nested(values['_parsed'], depth=10)
        return values

from pydantic import validator, parse_obj_as


class ChunkedData(BaseModel):
    chunks: List[str]

    @validator('chunks', pre=True)
    def split_data(cls, v):
        if isinstance(v, str):
            return [v[i:i + 1024] for i in range(0, len(v), 1024)]
        return v

    @root_validator
    def validate_chunks(cls, values):
        buffer = []
        for chunk in values['chunks']:
            buffer.append(parse_obj_as(DataChunk, chunk))
            if len(buffer) % 100 == 0:
                validate_buffer(buffer)
                buffer.clear()
        return values

class PipelineNode(BaseModel):
    input_schema: dict
    output_schema: dict
    next_nodes: List['PipelineNode'] = []

    @root_validator
    def validate_pipeline(cls, values):
        visited = set()

        def check_node(node):
            if id(node) in visited:
                return
            visited.add(id(node))
            if node.output_schema != node.next_nodes[0].input_schema:
                raise ValueError("节点schema不匹配")
            for n in node.next_nodes:
                check_node(n)

        check_node(values['self'])
        return values

class CompactModel(BaseModel):
    class Config:
        arbitrary_types_allowed = True
        copy_on_model_validation = 'none'

    @root_validator
    def optimize_memory(cls, values):
        for field in cls.__fields__:
            if isinstance(values[field], list):
                values[field] = tuple(values[field])
            elif isinstance(values[field], dict):
                values[field] = frozenset(values[field].items())
        return values

class GenerativeValidator(BaseModel):
    template: str
    dependencies: List['GenerativeValidator'] = []

    @root_validator
    def check_templates(cls, values):
        from jinja2 import Template, meta
        parsed = Template(values['template'])
        required_vars = meta.find_undeclared_variables(parsed)

        def collect_deps(node: 'GenerativeValidator', seen=None):
            seen = seen or set()
            if id(node) in seen:
                return set()
            seen.add(id(node))
            vars = meta.find_undeclared_variables(Template(node.template))
            for dep in node.dependencies:
                vars |= collect_deps(dep, seen)
            return vars

        available_vars = collect_deps(values['self'])
        if not required_vars.issubset(available_vars):
            missing = required_vars - available_vars
            raise ValueError(f"缺失模板变量: {missing}")
        return values

class DeltaValidator(BaseModel):
    base_version: int
    delta: dict
    _full_data: dict = None

    @root_validator
    def apply_deltas(cls, values):
        base = load_from_db(values['base_version'])
        values['_full_data'] = apply_delta(base, values['delta'])
        try:
            FullDataModel(**values['_full_data'])
        except ValidationError as e:
            raise ValueError(f"增量应用失败: {str(e)}")
        return values