Python3.11+Pydantic2.7大型模型高效数据收集优化方案

我正在使用Python 3.11和Pydantic 2.7框架，需要从一个体量庞大的模型中收集特定信息。目前有两种实现方式：

• 借助字段验证器在验证阶段收集数据
• 验证完成后遍历模型收集数据

我希望找到在解析/验证模型阶段更快收集数据的方法，此前尝试设置嵌套模型的实例变量、重写.model_validate()方法均未成功。以下是我的测试代码及测速逻辑：

import copy
from time import perf_counter as time
from typing import Any, ClassVar

from pydantic import BaseModel, field_validator


class Employees(BaseModel):
    name: str
    employees_names: ClassVar[list[str]] = []

    @field_validator('name')
    @classmethod
    def collect_employees_names(cls, value):
        cls.employees_names.append(value)

        return value


class EmployeesNative(BaseModel):
    name: str


class Manager(BaseModel):
    name: str
    employees: list[Employees]
    managers_names: ClassVar[list[str]] = []

    @field_validator('name')
    @classmethod
    def collect_managers_names(cls, value):
        cls.managers_names.append(value)

        return value


class ManagerNative(BaseModel):
    name: str
    employees: list[EmployeesNative]


class Results(BaseModel):
    managers: list[Manager]


class ResultsNative(BaseModel):
    managers: list[ManagerNative]


class Data(BaseModel):
    results: list[Results]


class DataNative(BaseModel):
    results: list[ResultsNative]


class Schema(BaseModel):
    data: Data


class SchemaNative(BaseModel):
    data: DataNative


man1 = {'name': 'max', 'employees': [{'name': 'Jorge'}, {'name': 'Sasha'}]}
man2 = {'name': 'kate', 'employees': [{'name': 'Eva'}, {'name': 'Mike'}, {'name': 'Riss'}]}
man3 = {'name': 'bub', 'employees': []}
man4 = {'name': 'slava', 'employees': [{'name': 'Martha'}]}

res1 = {'managers': [man1]}
res2 = {'managers': [man2, man3]}
res3 = {'managers': [man4]}


def parse_native(obj: dict[str, Any]) -> list[str]:
    """ Get all persons names in schema. """
    mod = SchemaNative.model_validate(obj)

    m = [m.name for res in mod.data.results for m in res.managers]
    e = [e.name for res in mod.data.results for m in res.managers for e in m.employees]
    r = m + e
    print('native: ', r)

    return r


def parse_validators(obj: dict[str, Any]) -> list[str]:
    """ Get all persons names in schema. """
    Schema.model_validate(obj)
    r = Manager.managers_names + Employees.employees_names
    Manager.managers_names = []
    Employees.employees_names = []
    print('validators: ', r)

    return r


def run_test(func, objs, comment='', reps=1, average=True, prec=.6):
    """ Speed checker. """
    total = 0
    for _ in range(reps):
        for e in objs:
            beg = time()
            func(e)
            end = time()
            total += (end - beg)
    t = total / reps if average else total
    print(f'{t:{prec}f} ({comment})')


# payload for tests
test_obj = {'data': {'results': [res1, res2, res3]}}
payload = tuple(copy.deepcopy(test_obj) for _ in range(1))

# number of repetitions
REPS = 1

run_test(func=parse_native, objs=payload, comment='No validators', reps=REPS)

run_test(func=parse_validators, objs=payload, comment='With validators', reps=REPS)

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