简介

pydantic-resolve 是一个基于 pydantic 的轻量级封装库， 它为 pydantic 和 dataclass 对象添加了 resolve 和 post 方法。

如果你曾经写过类似的代码， 并且觉得不满意， pydantic-resolve 可以派上用场。

story_ids = [s.id for s in stories]
tasks = await get_all_tasks_by_story_ids(story_ids)

story_tasks = defaultdict(list)

for task in tasks:
    story_tasks[task.story_id].append(task)

for story in stories:
    tasks = story_tasks.get(story.id, [])
    story.total_task_time = sum(task.time for task in tasks)
    story.total_done_tasks_time = sum(task.time for task in tasks if task.done)

上述代码的问题是将数据获取, 遍历和业务计算混淆在了一起.

pydantic-resolve 可以将他们拆分开来, 让开发专注在业务逻辑之上.

from pydantic_resolve import Resolver, LoaderDepend, build_list
from aiodataloader import DataLoader


# data fetching
class TaskLoader(DataLoader):
    async def batch_load_fn(self, story_ids):
        tasks = await get_all_tasks_by_story_ids(story_ids)
        return build_list(tasks, story_ids, lambda t: t.story_id)

# core business logics
class Story(Base.Story):
    tasks: List[Task] = []
    def resolve_tasks(self, loader=LoaderDepend(TaskLoader)):
        return loader.load(self.id)

    total_task_time: int = 0
    def post_total_task_time(self):
        return sum(task.time for task in self.tasks)

    total_done_task_time: int = 0
    def post_total_done_task_time(self):
        return sum(task.time for task in self.tasks if task.done)

# traversal and execute methods (runner)
await Resolver().resolve(stories)

同时他还能轻易地的扩展到更加复杂的场景, 比如额外增加一层 sprint 类, 如果用先前的代码, for 循环的缩进就已经会影响到代码的可读性了.

# data fetching
class TaskLoader(DataLoader):
    async def batch_load_fn(self, story_ids):
        tasks = await get_all_tasks_by_story_ids(story_ids)
        return build_list(tasks, story_ids, lambda t: t.story_id)

class StoryLoader(DataLoader):
    async def batch_load_fn(self, sprint_ids):
        stories = await get_all_stories_by_sprint_ids(sprint_ids)
        return build_list(stories, sprint_ids, lambda t: t.sprint_id)

# core business logic
class Story(Base.Story):
    tasks: List[Task] = []
    def resolve_tasks(self, loader=LoaderDepend(TaskLoader)):
        return loader.load(self.id)

    total_task_time: int = 0
    def post_total_task_time(self):
        return sum(task.time for task in self.tasks)

    total_done_task_time: int = 0
    def post_total_done_task_time(self):
        return sum(task.time for task in self.tasks if task.done)


class Sprint(Base.Sprint):
    stories: List[Story] = []
    def resolve_stories(self, loader=LoaderDepend(StoryLoader)):
        return loader.load(self.id)

    total_time: int = 0
    def post_total_time(self):
        return sum(story.total_task_time for story in self.stories)

    total_done_time: int = 0
    def post_total_done_time(self):
        return sum(story.total_done_task_time for story in self.stories)


# traversal and execute methods (runner)
await Resolver().resolve(sprints)

它可以在数据组装过程中， 降低获取和调整环节的代码复杂度， 使代码更加贴近 ER 模型， 更加可维护。

借助 pydantic 它可以像 GraphQL 一样用图的关系来描述数据结构， 也能够在获取数据的同时根据业务做调整。

它可以和 FastAPI 轻松合作， 在后端构建出前端友好的数据结构， 以 typescript sdk 的方式提供给前端。

在使用面向 ERD 的建模方式下， 它可以为你提供 3 ~ 5 倍的开发效率提升， 减少 50% 以上的代码量。

它为 pydantic 对象提供了 resolve 和 post 方法。

• resolve 通常用来获取数据
• post 可以在获取数据后做额外处理

from pydantic import BaseModel
from pydantic_resolve import Resolver
class Car(BaseModel):
    id: int
    name: str
    produced_by: str

class Child(BaseModel):
    id: int
    name: str

    cars: List[Car] = []
    async def resolve_cars(self):
        return await get_cars_by_child(self.id)

    description: str = ''
    def post_description(self):
        desc = ', '.join([c.name for c in self.cars])
        return f'{self.name} owns {len(self.cars)} cars, they are: {desc}'

children = await Resolver.resolve([
        Child(id=1, name="Titan"),
        Child(id=1, name="Siri")]
    )

当定义完对象方法， 并初始化好对象后， pydantic-resolve 内部会对数据做遍历， 执行这些方法来处理数据， 最终获取所有数据。

[
    Child(id=1, name="Titan", cars=[
        Car(id=1, name="Focus", produced_by="Ford")],
        description="Titan owns 1 cars, they are: Focus"
        ),
    Child(id=1, name="Siri", cars=[
        Car(id=3, name="Seal", produced_by="BYD")],
        description="Siri owns 1 cars, they are Seal")
]

对比面向过程的代码需要执行遍历和额外维护并发逻辑。

import asyncio

async def handle_child(child):
    cars = await get_cars()
    child.cars = cars

    cars_desc = '.'.join([c.name for c in cars])
    child.description = f'{child.name} owns {len(child.cars)} cars, they are: {car_desc}'

tasks = []
for child in children:
    tasks.append(handle(child))

await asyncio.gather(*tasks)

搭配 DataLoader， pydantic-resolve 可以避免多层获取数据时容易发生的 N+1 查询， 优化性能。

使用 DataLoader 还可以让定义的 class 片段在任意位置被复用。

除此以外它还提供了 expose 和 collector 机制为跨层的数据处理提供了便利。

安装

pip install pydantic-resolve

从 pydantic-resolve v1.11.0 开始， 将同时兼容 pydantic v1 和 v2。