使用 pydantic-resolve 简化 ER 模型到视图模型的流程

[tangkikodo]

借宝地推荐一个我开发，并且用着非常顺手的工具: pydantic-resolve

pydantic-resolve 是一个数据构建工具， 它和具体查询方式无关， 可以选择任意方式来为其提供数据

结合 ER 模型， 它有快速构建， 最大化复用查询，可灵活调整数据等优点。

pydantic-resolve 的内部执行的是支持异步的广度优先遍历， 从根节点遍历到子节点的过程中获取所需的关联数据

它同时还有一个遍历完成后的回溯阶段， 在回溯阶段， 默认树状数据的获取已经全部完成， 可以在这个阶段从子节点到根节点依次对数据做处理

处理的内容包括修改字段数据， 调整字段位置（把子节点收集到祖先节点）， 隐藏节点 （序列化时跳过）。

上述两个阶段对应的是数据构建过程中， 从源数据根据 ER 模型组合为业务数据，然后从业务数据变换为 UI 所需的视图数据的两步过程

ER 模型和业务概念

ER 模型的定义:

Entity-relationship model 用来描述特定知识领域中互相关联的相关事物， 由实体类型组成， 并且指定实体之间可能存在的关系

                     ┌───────────┐          
                     │           │          
                     │  project  │          
                     │           │          
                     └────┬┬─────┘          
                          ││                
       ┌──────────────────┘│   owns multiple
       │                   │                
       │                   │                
┌──────▼──────┐      ┌─────▼─────┐          
│             │      │           │          
│  timeline   │      │  budget   │          
│             │      │           │          
└─────────────┘      └───────────┘          

业务概念做什么用？ 业务概念约束了 ER 模型的使用方式， 明确指出了实体之间在怎样的业务场景下可以关联起来

以 project , budget, stage gate 这三个 entity 为例子

在 budget 的业务概念中， 用户只需要关心 project 和 budget，而在 timeline 的业务概念中， 用户只关心 project 和 timeline ，并不需要知道 budget 的存在

这个对开发来说也是一样的， 开发在处理 budget 相关业务时， 不应该去关心 timeline 相关的信息

打个比方，我有个 entity 的大池子，每次我会根据实际业务要求，将所需的 entity 捞出来，接着根据业务所需的 relationship 组合为业务数据

业务所需的 relationship , 即使都是 project 到 budget ，定义也可能不同， 比如

• project 要关联一个时间段内的 budget ，
• project 要关联到未来计划的 budget
• project 关联数据异常的 budget

这里面 relationship 的参数都是不同的

如何定义 ER 模型

DB

这是我们日常使用最多的方式

但用数据库来定义 ER 模型， 和面向业务描述的 ER 模型， 两者之间存在许多区别

DB ER 模型是个 “贫血”模型， 用它定义关联， 会将多种业务概念混杂在一起， 用简化的代码来展示就类似

class BaseProject
    id: int
    name: str
    
    budgets: list[Budget]
    timelines: list[Timeline]

在阅读的过程中很难区分出 budgets 和 timelines 是分属于两种业务概念的， 写查询的时候也可能因为业务不熟悉错误关联了其他业务的数据。

如果换一种表达方式， 使用继承的方式， 就能清晰区分出来两业务概念了

class ProjectOfBudget(Project):
    budgets: list[Budget]
    
class ProjectOfTimeline(Project):
    timelines: list[Timeline]

它能清晰表达出 Project + budgets 是一组业务概念的成员， 但 DB/ORM 属于具体实现层， 所以将业务概念混合进去并不是很合理。

DB 定义的 ER 模型， 应该追随业务的 ER 模型， 而不是反过来变成主角

比如多对多的概念在 DB 中会使用中间表来实现， 而业务上的多对多会根据入口数据的不同， 简化成一对多的情况

比如从单个 budget 的角度可以描述它归属于多个 project ， 但此时并不需要关心 DB 实现的时候是不是引入了中间表。

使用 DB 来定义 ER 模型的问题就是这些具体实现所包含的 “杂音”， 在业务 ER 模型中， 我只关心关联的形式， 不用关心具体关联的手段。

┌──────────┐
│          │
│  budget  │
│          │
└─────┬────┘
      │     
      │     owns many
      │     
      │     
┌─────▼────┐
│          │
│  project │
│          │
└──────────┘

pydantic-resolve

使用 pydantic 的时候， 因为无需关心具体实现， 所以可以更加专注在 ER 模型上

我们可以先将所有的 entity 定义出来

class BaseProject(BaseModel)
    id: int
    name: str

class BaseTimeline(BaseModel):
    id: int
    name: str
    
class BaseBudget(BaseModel):
    id: int
    name: str

接着， 根据业务概念， 来选取所需的 entity ， 并定义好 relationship 数据如何获取

比如对应 budget 业务， 挑选出 Project 和 Budget

class Project(BaseProject):
    budgets: list[BaseBudget] = []

然后根据业务细节（获取一段时间内的 budget ）， 定义 dataloader 的具体实现：

class Project(BaseProject):
    # projects with budget, budget should support filter by date range
    
    budgets: list[BaseBudget] = []  
    def resolve_budgets(self, loader=LoaderDepends(BudgetByRangeLoader)):
        return loader.load(self.id)
        
projects = await get_projects()
projects = await Resolver(loader_params: {
    BudgetByRangeLoader: { 'start': '2021-12-21', 'end': '2022-12-21'
}).resolve(projects)

此处的 BudgetByRangeLoader 就交给 DB 或者其他 API 去实现了， 而且这样的查询写起来会很简单， 可复用性也更高 （简单的 select where in)

这种做法隐藏的缺点是查询总时间会变长， 因为获取 projects 和 获取 budget 的两次查询是无法并发的

当然， 如果需求稳定下来之后， 想要优化也很容易，移除 resolve_budget ， 直接用 DB 查询整个 project 和 budget 的数据

resolve_budgets 只是一种可用的查询手段， 更重要的是 pydantic 定义好的业务概念。

业务模型转换为视图模型

DB 处理转换

DB 确实有能力处理视图数据， 而且理论上性能是最好的

但直接用 DB 查询生成视图数据， 查询容易写的很复杂， 代码的可阅读和可维护性也是很重要的指标， 许多项目早期调整很多，DB 的查询修改起来会比较费劲

查询中既有 join 做关联， 又有 where 做过滤， 还有 select 语句中做数据转换， 在处理单表或者少数表的时候也许还行

可是一旦表增多， 就容易难以维护

另外构建多层嵌套结构的 sql 查询往往会涉及到拼装 json 格式， 比如

    select_query = (
        select(
            (
                posts.c.id,
                posts.c.slug,
                posts.c.title,
                func.json_build_object(
                   text("'id', profiles.id"),
                   text("'first_name', profiles.first_name"),
                   text("'last_name', profiles.last_name"),
                   text("'username', profiles.username"),
                ).label("creator"),
            )
        )
        .select_from(posts.join(profiles, posts.c.owner_id == profiles.c.id))
        .where(posts.c.owner_id == creator_id)
        .limit(limit)
        .offset(offset)
        .group_by(
            posts.c.id,
            posts.c.type,
            posts.c.slug,
            posts.c.title,
            profiles.c.id,
            profiles.c.first_name,
            profiles.c.last_name,
            profiles.c.username,
            profiles.c.avatar,
        )
        .order_by(
            desc(coalesce(posts.c.updated_at, posts.c.published_at, posts.c.created_at))
        )
    )

这个是两层的例子， 如果要再增加一层， 查询语句就会更加复杂

ORM 可以构建多层嵌套数据， 但是如果想在 relationship 的过程中加入额外约束条件一般来说还是挺麻烦的

使用代码处理转换

这种就非常常见了， 比如使用 ORM 获取到了树状数据， 然后使用 pandas 或者手写转换代码

比如为 story 添加 total_count

for project in projects:
    total = 0
    total_invalid = 0
    for budget in project.budgets:
        total += budget.value
        if budget.is_invalid:
            total += budget.value

缺点还是可读性， 如果层级加深就会快速恶化

使用 pydantic-resolve 处理转换

在上文的基础上， 我们只要为 Project 添加两个额外字段，无需自己对 projects 做额外的展开遍历

class Project(BaseProject):
    # projects with budget, budget should support filter by date range
    
    budgets: list[BaseBudget] = []  
    def resolve_budgets(self, loader=LoaderDepends(BudgetByRangeLoader)):
        return loader.load(self.id)
        
    total: int = 0
    def post_total(self):
        return sum([b for b in self.budgets])
    
    total_invalid = 0
    def post_total_invalid(self):
        return sum([b for b in self.budgets if b.is_invalid])

如果觉得两次 for loop 有点不划算， 也可以使用 post_default_handler 来优化

class Project(BaseProject):
    # projects with budget, budget should support filter by date range
    
    budgets: list[BaseBudget] = []  
    def resolve_budgets(self, loader=LoaderDepends(BudgetByRangeLoader)):
        return loader.load(self.id)
        
    total: int = 0
    total_invalid = 0
    
    def post_default_handler(self):
        for budget in project.budgets:
        self.total += budget.value
        if budget.is_invalid:
            self.total += budget.value

post 阶段还有许多功能 ，比如将子孙字段聚合到祖先节点，或者在 post 阶段额外执行异步任务来获取数据， 就不一一赘述了。

reference： https://allmonday.github.io/pydantic-resolve/zh/v2/expose_and_collect/

总结

pydantic-resolve 将注意力放在了业务模型和业务概念上， 通过简洁的语法定义， 将 ER 模型和业务概念描述出来

class Project(BaseProject):
    # business: projects with budget, budget should support filter by date range
    
    budgets: list[BaseBudget] = []
    def resolve_budgets(self, loader=LoaderDepends(BudgetByRangeLoader)):
        return loader.load(self.id)

同时借助 dataloader 来提供一套最基础的实现框架， 只要满足 batch 查询就能将数据关联上去， 可以是 DB 查询， 也可以是 RPC 调用

这种方式对架构调整比较友好，即使具体实现改变了，pydantic 这层定义也无需调整

在回溯阶段，pydantic-resolve 提供充分的灵活度， 支持数据调整， 数据移动， 数据隐藏等需求， 满足各种视图构建过程中的需要。