让 ER model 始终清晰可见
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 # 交互式分析数据结构通过声明式描述数据结构，自动构建多层嵌套的 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" }
}
]- Example 1: 多层嵌套结构的构建
- Example 2: Loader 的进阶用法
- Example 3: 跨层级数据获取
- Example 4: 每层数据的后处理
- Example 5: 利用 Context 和 Schema 实现复用
- Example 6: 挑选字段
- Example 7: 直接操作 Loader 实例
- 更灵活的测试: 用 service 测试代替 API 测试
- 其他: 和 GraphQL 比较
- 使用 openapi codegen 和前端集成
构建面向视图的数据时，不可避免会出现数据组装需求：
{
"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 通过声明式描述数据结构是一个好的方向：
{
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可以交互式分析项目的数据结构
完.