GitHub - allmonday/rapid-development-pattern: build whatever you want by pydantic-resolve

面向组合的 API 开发模式

让 ER model 始终清晰可见

english version

generated by fastapi-voyager

简介

使用面向组合的开发模式，结合 pydantic-resolve，写出更容易维护、更好分析的业务逻辑。

核心思想

传统数据组装的问题：

# 命令式 - 繁琐且难以维护
for team in teams:
    for member in team.members:
        member.tasks = get_tasks_by_member(member.id)

组合模式的方式：

# 声明式 - 清晰且自动批量加载
class TaskResponse(BaseModel):
    user: Optional[User] = None
    def resolve_user(self, loader=Loader(user_batch_loader)):
        return loader.load(self.owner_id)

class MemberResponse(BaseModel):
    tasks: list[TaskResponse] = []
    def resolve_tasks(self, loader=Loader(member_to_tasks_loader)):
        return loader.load(self.id)

# 使用 Resolver 自动解析
result = await Resolver().resolve(members)

快速开始

运行项目

python -m venv venv
source venv/bin/activate
pip install -r requirement.txt
uvicorn src.main:app --port=8000 --reload
# http://localhost:8000/docs
# http://localhost:8000/voyager  # 交互式分析数据结构

示例：Mini JIRA

通过声明式描述数据结构，自动构建多层嵌套的 API 响应：

from typing import Optional
from pydantic_resolve import LoaderDepend as LD

class Sample1TaskDetail(ts.Task):
    user: Optional[us.User] = None
    def resolve_user(self, loader=LD(ul.user_batch_loader)):
        return loader.load(self.owner_id)

class Sample1StoryDetail(ss.Story):
    tasks: list[Sample1TaskDetail] = []
    def resolve_tasks(self, loader=LD(tl.story_to_task_loader)):
        return loader.load(self.id)

    owner: Optional[us.User] = None
    def resolve_owner(self, loader=LD(ul.user_batch_loader)):
        return loader.load(self.owner_id)

@route.get('/stories-with-detail', response_model=List[Sample1StoryDetail])
async def get_stories_with_detail(session: AsyncSession = Depends(db.get_session)):
    stories = await sq.get_stories(session)
    stories = [Sample1StoryDetail.model_validate(t) for t in stories]
    stories = await Resolver().resolve(stories)
    return stories

输出：

[
  {
    "id": 1,
    "name": "deliver a MVP",
    "tasks": [
      {
        "id": 1,
        "name": "mvp tech design",
        "user": { "id": 2, "name": "Eric" }
      }
    ],
    "owner": { "id": 1, "name": "John" }
  }
]

功能示例

为什么需要组合模式？

传统方式的困境

构建面向视图的数据时，不可避免会出现数据组装需求：

{
  "team": "a",
  "members": [
    {
      "name": "kikodo",
      "tasks": [{ "name": "complete tutorial" }]
    }
  ]
}

传统做法是手动循环拼接：

task_map = group_by_member_id(tasks)
member_map = group_by_team_id(members)

for m in members:
    m.tasks = task_map[m.id]
for t in teams:
    t.members = member_map[t.id]

问题：

过程式代码对调整和阅读不友好
循环和拼接产生不通用、不易维护的代码
添加和修改字段很麻烦
分层困难（放在 controller/service/model 都有问题）

GraphQL 的启示与局限

GraphQL 通过声明式描述数据结构是一个好的方向：

{
  project(name: "GraphQL") {
    tagline
  }
}

但 GraphQL 也有问题：

无法描述尺寸不确定的递归结构
复杂查询的性能优化困难
数据后期处理不便
架构侵入较大

组合模式的优势

省去 GraphQL 的查询部分，保留其声明式描述的核心思想：

from pydantic import BaseModel
from pydantic_resolve import Resolver

class HelloView(BaseModel):
    hello: str = ''
    def resolve_hello(self, context):
        return f"Hello {context['first_name']}"

    goodbye: str = ''
    def resolve_goodbye(self):
        return 'See ya'

    def post_goodbye(self):
        return 'See ya soon'  # 数据后处理

result = await Resolver(context={'first_name': 'kikodo'}).resolve(HelloView())

核心理念：把大而全的单一查询入口，替换成一个个小巧灵活的定制化 schema 描述。

架构设计

核心概念

定义视图结构 schema（从根数据向下扩展）
获取根数据（树干），转换成 schema
Resolver 遍历解析所有数据（树枝、树叶）

Resolver 过程包含：

Forward fetch：向下获取关联数据
Backward change：数据后处理（post 方法）
Exclude fields：字段筛选

分层设计

Service 层：提供稳定的业务 query、mutation、schema、loader
Router 层：声明面向组合的视图 schema，组合 service 提供的数据

service (稳定)
  - query: 业务查询（主数据）
  - loader: 关联数据（可扩展）
  - schema: 业务类型

router (灵活)
  - 组合 service 的 schema
  - 声明视图结构
  - Resolver 自动解析

优势

查询层面：

声明式描述数据，直观易修改
简化根数据查询，避免复杂 SQL
支持 N+1 查询优化（DataLoader）

调整层面：

每层都有后处理能力（post 方法）
可挑选字段、隐藏字段
直接满足前端所需复杂结构

性能层面：

避免 N+1 查询
对优化友好

协作层面：

OpenAPI 自动生成 SDK
TypeScript 类型安全
前后端调整变得简单

测试优势

只要 service 层有充分的测试覆盖，router 层的组合功能基本不需要测试。

数据源可靠 + 组合过程可靠 = 视图数据可靠

总结

组合模式的核心价值：分离业务中的稳定和不稳定的部分。

Service 保持稳定：

query 专注主数据查询
loader 提供关联数据
对外暴露清晰的接口

Router 灵活组合：

按需继承 service schema
每个接口独立优化
快速响应需求变化

这种模式让核心业务逻辑的可维护性提升，测试更容易覆盖，为架构演进（如单体→微服务）保留弹性。

注意：访问 /voyager 可以交互式分析项目的数据结构

完.