基于 Pydantic-Resolve 和 FastAPI-Voyager 的 Clean Architecture 实践
篇幅较长无法粘贴全文，原文链接：
https://github.com/allmonday/A-Python-web-development-methodology-for-complex-business-scenarios/blob/main/README.zh.md
一套面向复杂业务场景的 Python Web 开发方法论
目录
1. 背景与问题
1.1 当前主流做法及其痛点
在 Python Web 开发中，处理复杂业务场景时，开发者通常采用以下几种模式：
模式一：直接使用 ORM （如 SQLAlchemy ）
@router.get("/teams/{team_id}", response_model=TeamDetail)
async def get_team(team_id: int, session: AsyncSession = Depends(get_session)):
# 获取团队基本信息
team = await session.get(Team, team_id)
# 获取 Sprint 列表
sprints = await session.execute(
select(Sprint).where(Sprint.team_id == team_id)
)
team.sprints = sprints.scalars().all()
# 获取每个 Sprint 的 Story
for sprint in team.sprints:
stories = await session.execute(
select(Story).where(Story.sprint_id == sprint.id)
)
sprint.stories = stories.scalars().all()
# 获取每个 Story 的 Task
for story in sprint.stories:
tasks = await session.execute(
select(Task).where(Task.story_id == story.id)
)
story.tasks = tasks.scalars().all()
# 获取每个 Task 的负责人
for task in story.tasks:
task.owner = await session.get(User, task.owner_id)
return team
这种做法在简单场景下确实很直观，能够快速上手。ORM 的类型安全特性也能在编译时发现一些错误，而且与数据库表结构的一一对应关系让代码容易理解。但当我们面对真正的业务场景时，这种方式的缺陷很快就暴露出来了。
最致命的问题是 N+1 查询。虽然代码看起来很清晰，但执行时会产生大量的数据库查询。每当我们访问一个关联关系时，ORM 就会发起一次新的查询。在深层嵌套的情况下，查询数量会呈指数级增长。更糟糕的是，这种性能问题在开发阶段不容易发现，只有当数据量积累到一定程度后才会显现出来，那时候往往已经太晚了。
代码的组织方式也是个问题。数据获取的逻辑散落在各个嵌套的循环中，业务逻辑和数据获取逻辑混在一起，难以阅读和维护。当需要修改业务规则时，开发者不得不在复杂的嵌套结构中寻找修改点，很容易引入新的 bug 。性能更是不可控，随着数据量的增长，查询效率会急剧下降，而这些性能瓶颈很难在代码层面直接观察到。
此外，相似的数据获取逻辑会在多个 API 中重复出现，导致大量代码冗余。当一个 API 需要获取"团队及其 Sprint"，另一个 API 需要"团队及其成员"时，即使它们的查询逻辑非常相似，也不得不重复编写。这违反了 DRY （ Don't Repeat Yourself ）原则，增加了维护成本。
模式二：使用 ORM 的 Eager Loading
@router.get("/teams/{team_id}", response_model=TeamDetail)
async def get_team(team_id: int, session: AsyncSession = Depends(get_session)):
# 使用 joinedload 预加载关联数据
result = await session.execute(
select(Team)
.options(
joinedload(Team.sprints)
.joinedload(Sprint.stories)
.joinedload(Story.tasks)
.joinedload(Task.owner)
)
.where(Team.id == team_id)
)
return result.scalar_one()
为了解决 N+1 查询问题，ORM 提供了 Eager Loading 机制，让我们可以通过 joinedload、selectinload 等方式预先加载关联数据。代码变得更简洁了，性能问题也得到了缓解。但这种方案也带来了新的挑战。
最明显的问题是笛卡尔积。当我们使用多层 JOIN 预加载关联数据时，数据库返回的数据量会急剧膨胀。比如一个团队有 10 个 Sprint ，每个 Sprint 有 10 个 Story ，每个 Story 有 10 个 Task ，那么 JOIN 的结果集会包含 1000 行数据，即使每行的数据量不大，也会给网络传输和内存占用带来压力。
更严重的问题是灵活性差。Eager Loading 的策略是在代码中硬编码的，所有使用同一个 Model 的 API 都会执行相同的预加载逻辑。但不同的 API 往往需要不同的数据。比如一个 API 只需要团队的基本信息，另一个 API 需要团队的 Sprint ，还有一个 API 需要团队的成员。如果统一使用 Eager Loading 加载所有关联数据，就会出现过度获取的问题，前端不需要的数据也被查询和传输了，浪费了资源。
配置 Eager Loading 本身就很复杂。开发者需要理解 lazy、joinedload、selectinload、subquery 等多种加载策略的区别，知道什么时候用哪一种，以及它们各自会有什么副作用。这种配置错误很容易导致性能问题或意外的数据加载行为。而且，这种"一刀切"的配置方式意味着所有 API 都使用相同的加载策略，无法针对特定场景进行优化。
模式三：手动组装数据
@router.get("/teams/{team_id}", response_model=TeamDetail)
async def get_team(team_id: int, session: AsyncSession = Depends(get_session)):
# 1. 批量获取所有需要的数据
team = await session.get(Team, team_id)
sprints_result = await session.execute(
select(Sprint).where(Sprint.team_id == team_id)
)
sprint_ids = [s.id for s in sprints_result.scalars().all()]
stories_result = await session.execute(
select(Story).where(Story.sprint_id.in_(sprint_ids))
)
story_ids = [s.id for s in stories_result.scalars().all()]
tasks_result = await session.execute(
select(Task).where(Story.id.in_(story_ids))
)
tasks = tasks_result.scalars().all()
owner_ids = list(set(t.owner_id for t in tasks))
owners_result = await session.execute(
select(User).where(User.id.in_(owner_ids))
)
owners = {u.id: u for u in owners_result.scalars().all()}
# 2. 手动组装数据结构
sprint_dict = {s.id: s for s in sprints_result.scalars().all()}
story_dict = {s.id: s for s in stories_result.scalars().all()}
for story in story_dict.values():
story.tasks = [t for t in tasks if t.story_id == story.id]
for task in story.tasks:
task.owner = owners.get(task.owner_id)
for sprint in sprint_dict.values():
sprint.stories = [s for s in story_dict.values() if s.sprint_id == sprint.id]
team.sprints = list(sprint_dict.values())
return team
为了获得最优的性能和精确的数据控制，有经验的开发者会选择手动组装数据。这种方式完全掌控查询逻辑，可以精确控制每个查询的 SQL 语句，避免不必要的数据库访问。通过批量查询和智能的数据组装，可以获得最佳的性能，而且没有冗余数据。
但这种方式的代价是代码变得非常冗长。如上面的例子所示，为了获取一个团队的完整信息，我们需要编写多个查询，手动构建数据字典，然后通过嵌套循环组装数据。代码的长度和复杂度都大幅增加，而真正表达业务逻辑的代码反而被淹没在数据组装的细节中。
更容易出错也是个大问题。手动组装数据涉及到大量的索引操作和循环嵌套，很容易出现索引错误、空指针引用等 bug 。而且这些错误往往只有在运行时、特定数据条件下才会暴露，难以在开发阶段发现。
维护成本更是高昂。当业务规则发生变化时（比如需要添加一个新的关联关系），开发者需要在所有相关的 API 中修改数据组装逻辑。如果遗漏了某个地方，就会导致数据不一致。而且，相似的数据组装逻辑会在多个 API 中重复出现，违反了 DRY 原则。
最根本的问题是，这种代码已经变成了纯粹的数据搬运工，看不出任何业务意图。代码中充满了字典操作、循环嵌套、索引查找，而这些都是技术细节，与业务需求毫无关系。新加入的团队成员很难从这些代码中理解业务逻辑，业务知识的传递变得异常困难。
模式四：使用 GraphQL
type Query {
team(id: ID!): Team
}
type Team {
id: ID!
name: String!
sprints: [Sprint!]!
}
type Sprint {
id: ID!
name: String!
stories: [Story!]!
}
type Story {
id: ID!
name: String!
tasks: [Task!]!
}
type Task {
id: ID!
name: String!
owner: User!
}
GraphQL 确实是一个很有吸引力的方案。前端可以按需获取数据，需要什么字段就查什么字段，不会有过度获取的问题。它提供了类型安全的查询接口，而且通过 DataLoader 可以自动解决 N+1 查询问题。这些特性让 GraphQL 在前端开发中广受欢迎。
但 GraphQL 的学习曲线非常陡峭。开发者需要学习全新的查询语言、Schema 定义、Resolver 编写、DataLoader 配置等一堆概念，这与 REST API 的直观性形成了鲜明对比。更麻烦的是，GraphQL 的过度灵活性给后端带来了巨大的挑战。前端可以构造任意复杂的查询，有些查询甚至可能是开发者没有想到过的，这导致后端很难进行针对性的优化。当一个查询嵌套了 10 层，返回了数百万条数据时，数据库和服务器都会面临巨大的压力。
调试 GraphQL API 也比调试 REST API 复杂得多。当一个 GraphQL 查询出错时，错误信息往往很难定位到具体的问题源头。而且 GraphQL 需要额外的服务器和工具链支持，无法直接利用现有的 FastAPI 生态系统。比如 FastAPI 的依赖注入、中间件、自动文档生成等特性，在 GraphQL 中都无法直接使用。
还有一个更深层次的问题是 ERD 和用例的界限模糊。GraphQL 的 Schema 同时扮演了实体模型和查询接口两个角色。当我们设计一个 GraphQL Schema 时，很难确定应该按照实体来组织（一个 Type 对应一个数据库表），还是按照用例来组织（不同的业务场景需要不同的字段）。这导致最佳实践不清晰，不同的项目、不同的开发者可能有完全不同的组织方式。
而且随着业务增长，所有的用例都会堆砌在同一个 Schema 中，导致 Schema 膨胀，难以维护。权限控制也变得异常复杂。不同的 API 端点可能有不同的权限要求，但它们可能都查询同一个实体（比如 User ），在 GraphQL 中很难针对不同的查询场景应用不同的权限规则。
1.2 问题根源分析
上面我们探讨的所有模式，虽然表面上的问题各不相同，但它们的核心困境其实是一致的。
问题 1：业务模型与数据模型混淆
# SQLAlchemy ORM 同时扮演两个角色：
# 1. 数据模型（如何存储）
# 2. 业务模型（业务概念）
class Team(Base):
__tablename__ = 'teams'
id = Column(Integer, primary_key=True)
name = Column(String)
# 这是数据库的外键关系，还是业务关系？
sprints = relationship("Sprint", back_populates="team")
在传统的 ORM 开发中，业务模型和数据模型是混在一起的。看看这个例子，Team 类既表达了业务概念（团队是什么），又承载了数据模型的细节（如何在数据库中存储）。当我们在 sprints 字段上定义 relationship 时，这到底是在描述一个业务关系（团队有多个 Sprint ），还是在声明一个数据库外键约束？这种模糊性会导致很多问题。
数据库的设计约束会直接影响我们的业务建模。比如，如果数据库中的 teams 表没有直接到 users 的外键，而是通过中间表 team_members 关联，那么在 ORM 中我们也必须通过这个中间表来定义关系。这意味着业务模型被迫适应数据库的实现细节，而不是反过来。
更严重的是，这种方式无法表达跨库、跨服务的业务关系。现代系统中，数据可能分布在不同的数据库中，甚至存储在外部服务里。比如用户的基本信息在 PostgreSQL ，而用户的偏好设置在 MongoDB ，用户的实时状态在 Redis 中。ORM 的 relationship 无法跨越这些边界，业务模型因此被限制在了单一数据库的范围内。
问题 2：依赖方向错误
传统架构的依赖方向：
┌─────────────┐
│ API Layer │ ← 依赖于
└──────┬──────┘
│
↓
┌─────────────┐
│ ORM Models │ ← 依赖于
└──────┬──────┘
│
↓
┌─────────────┐
│ Database │
└─────────────┘
问题：业务规则依赖于数据库实现！
这违反了 Clean Architecture 的依赖规则。正确的依赖关系应该是：业务规则最稳定，不依赖任何外层；数据库是实现细节，应该依赖业务规则；当数据库变化时，业务规则不应该受影响。但传统架构的依赖方向恰恰相反，业务规则被数据库的实现细节所绑架。
问题 3：缺少业务关系的显式声明
# 传统方式：业务关系隐藏在查询中
async def get_team_tasks(team_id: int):
# "团队的任务"这个业务概念隐藏在 SQL WHERE 中
result = await session.execute(
select(Task)
.join(Sprint, Sprint.id == Task.sprint_id)
.where(Sprint.team_id == team_id)
)
return result.scalars().all()
业务关系没有被显式声明出来，这是个很隐蔽但危害很大的问题。看看这个例子，"团队的任务"是一个清晰的业务概念，但这个概念被隐藏在 SQL 的 JOIN 和 WHERE 子句中。新加入团队的成员需要阅读大量代码才能理解系统中有哪些业务关系，这些关系是如何定义的。更糟糕的是，没有自动化的方式来检查业务关系的一致性。当需求变化需要修改某个关系时，开发者很难找到所有相关的代码，很容易遗漏某个地方，导致业务逻辑的不一致。
问题 4：中间表的技术暴露
在 SQLAlchemy ORM 中，多对多关系需要显式定义中间表，这导致技术细节泄漏到业务层。
# SQLAlchemy ORM：必须定义中间表
class Team(Base):
__tablename__ = 'teams'
id = Column(Integer, primary_key=True)
name = Column(String)
# ORM relationship 需要指定中间表
members = relationship("User",
secondary="team_members", # 必须指定中间表
back_populates="teams")
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
teams = relationship("Team",
secondary="team_members", # 必须指定中间表
back_populates="members")
# 中间表（技术实现细节）
class TeamMember(Base):
__tablename__ = 'team_members'
team_id = Column(Integer, ForeignKey('teams.id'), primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'), primary_key=True)
role = Column(String) # 可能还有额外字段
# 查询时需要关心中间表的存在
@router.get("/teams/{team_id}")
async def get_team_members(team_id: int, session: AsyncSession):
# 必须通过中间表查询
result = await session.execute(
select(User)
.join(TeamMember, TeamMember.user_id == User.id) # 中间表暴露
.where(TeamMember.team_id == team_id)
)
return result.scalars().all()
这个问题的根源在于，ORM 的多对多关系需要显式定义中间表，这导致技术细节直接泄漏到业务层代码中。业务代码必须知道 team_members 中间表的存在，查询时也需要显式地 join 这个中间表。这增加了代码复杂度，更重要的是，业务逻辑被数据库的实现细节所绑架。
更深层的问题是业务语义变得模糊。TeamMember 到底是一个有意义的业务概念，还是纯粹的技术实现？如果中间表还有额外的字段（比如 role 表示用户在团队中的角色，joined_at 表示加入时间），这些字段应该被建模为独立的实体吗？不同的开发者可能给出不同的答案，缺乏统一的指导原则。
数据组装也因此变得复杂。查询"团队的所有成员"需要 join 中间表，查询"用户所属的团队"也需要 join 中间表。所有涉及多对多关系的查询都变得冗长和难以理解。当业务规则要求"获取用户在所有团队中的角色"时，情况就更加复杂了。这些技术细节让业务逻辑的实现变得异常沉重。
对比：Pydantic-Resolve ERD 的方式
# ERD：业务概念清晰，无需关心中间表
class TeamEntity(BaseModel, BaseEntity):
"""团队实体 - 业务概念"""
__relationships__ = [
# 直接表达"团队有多个成员"的业务关系
Relationship(
field='id',
target_kls=list[UserEntity],
loader=team_to_users_loader # loader 内部处理中间表
),
]
id: int
name: str
class UserEntity(BaseModel, BaseEntity):
"""用户实体 - 业务概念"""
__relationships__ = [
# 直接表达"用户属于多个团队"的业务关系
Relationship(
field='id',
target_kls=list[TeamEntity],
loader=user_to_teams_loader
),
]
id: int
name: str
# Loader 实现细节：中间表只在这里出现
async def team_to_users_loader(team_ids: list[int]):
"""加载团队成员 - 内部处理中间表"""
async with get_session() as session:
# 只有这里需要知道中间表的存在
result = await session.execute(
select(User)
.join(TeamMember, TeamMember.user_id == User.id)
.where(TeamMember.team_id.in_(team_ids))
)
users = result.scalars().all()
# 构建映射
users_by_team = {}
for user in users:
for tm in user.team_memberships:
if tm.team_id not in users_by_team:
users_by_team[tm.team_id] = []
users_by_team[tm.team_id].append(user)
return [users_by_team.get(tid, []) for tid in team_ids]
关键差异：
| 维度 |
SQLAlchemy ORM |
Pydantic-Resolve ERD |
| 中间表位置 |
暴露在业务层 |
隐藏在 loader 实现中 |
| 业务语义 |
技术关系 (secondary) |
业务关系 (团队包含成员) |
| 查询代码 |
需要 join 中间表 |
loader.load(team_id) |
| 代码位置 |
分散在多处 |
集中在 loader |
| 测试 |
依赖数据库表结构 |
可 mock loader |
架构优势：
传统方式：
Team → TeamMember (中间表) → User
业务层需要知道中间表的存在
Pydantic-Resolve 方式：
Team → User (业务关系)
中间表是数据层的实现细节，业务层不关心
这意味着：
业务模型纯净：Team 和 User 的关系直接表达业务语义
技术细节封装：中间表的存在被封装在 loader 中
灵活的存储策略：
- 数据库可以用中间表实现
- 也可以用 JSON 字段存储
- 甚至可以是外部服务（如 LDAP ）
- 业务层代码无需修改
易于理解：新人看到 ERD 就能理解业务关系，不需要先学习数据库设计
2. Clean Architecture 思想
2.1 核心原则
Clean Architecture 由 Robert C. Martin (Uncle Bob) 提出，核心思想是：
"Software architecture is the art of drawing lines that I call boundaries."
软件架构的艺术在于画界线。
原则 1：依赖规则
外层依赖内层，内层不依赖外层。
↓ 依赖方向
┌─────────────────────┐
│ Frameworks & │ 外层
│ Drivers │ (实现细节)
├─────────────────────┤
│ Interface │
│ Adapters │
├─────────────────────┤
│ Use Cases │
│ (Application) │
├─────────────────────┤
│ Entities │ 内层
│ (Business Rules) │ (核心)
└─────────────────────┘
遵循依赖规则有几个关键点需要注意。首先，内层不知道外层的存在，这意味着核心业务逻辑不依赖于任何框架、数据库或 UI 的细节。其次，内层不包含外层的信息，比如业务规则不应该知道数据是用 PostgreSQL 还是 MongoDB 存储的。最后，外层的实现可以随时替换而不影响内层，这意味着我们可以从 SQLAlchemy 切换到 MongoDB ，或者从 FastAPI 切换到 Django ，而业务逻辑代码无需修改。
原则 2：业务规则独立
# ❌ 错误：业务规则依赖数据库
class Task:
def calculate_priority(self, session):
# 业务逻辑被数据库实现细节污染
if self.assignee_id in session.query(TeamMember).filter_by(role='lead'):
return 'high'
# ✅ 正确：业务规则独立
class Task:
def calculate_priority(self, assignee_roles):
# 业务逻辑只依赖业务概念
if 'lead' in assignee_roles:
return 'high'
原则 3：跨边界的数据传递
# 内层定义数据结构
class TaskEntity(BaseModel):
id: int
name: str
assignee_id: int
# 外层负责转换
def task_entity_to_orm(entity: TaskEntity) -> Task:
return Task(
id=entity.id,
name=entity.name,
assignee_id=entity.assignee_id
)
2.2 依赖规则
在 Web 开发中，依赖规则可以这样理解：
┌────────────────────────────────────────────────────┐
│ Presentation Layer (外层) │
│ - FastAPI Routes │
│ - Request/Response Models │
│ - 依赖: Application Layer │
└────────────────────────────────────────────────────┘
↓
┌────────────────────────────────────────────────────┐
│ Application Layer (Use Cases) │
│ - 业务用例（获取用户、创建订单） │
│ - 依赖: Domain Layer │
└────────────────────────────────────────────────────┘
↓
┌────────────────────────────────────────────────────┐
│ Domain Layer (内层) │
│ - Entities (业务实体) │
│ - Business Rules (业务规则) │
│ - Value Objects (值对象) │
│ - 不依赖任何外层 │
└────────────────────────────────────────────────────┘
↓
┌────────────────────────────────────────────────────┐
│ Infrastructure Layer (最外层) │
│ - Database (SQLAlchemy) │
│ - External Services │
│ - File System │
└────────────────────────────────────────────────────┘
关键洞察：
- Entities 不应该知道 SQLAlchemy 的存在
- Business Rules 不应该知道数据库表结构
- Use Cases 不应该知道 HTTP 协议的细节
2.3 在 Web 开发中的应用
传统架构的问题
# 传统方式：所有层次耦合
# Domain Layer (应该独立，但实际上依赖了 ORM)
class User(Base): # ← SQLAlchemy Base
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
# Application Layer (应该只依赖 Domain ，但直接使用了 ORM)
async def create_user(data: dict, session: AsyncSession):
user = User(**data) # ← 直接使用 ORM Model
session.add(user)
await session.commit()
# Presentation Layer
@router.post("/users")
async def api_create_user(data: dict, session=Depends(get_session)):
return await create_user(data, session) # ← 暴露了数据库细节
这段代码暴露了传统架构的核心问题。SQLAlchemy 虽然建立了对象关系映射（ ORM ），让数据库表可以通过 Python 对象来操作，但这种映射关系过于紧密。ORM Model 既承担了数据持久化的职责，又要表达业务概念，导致对象无法自由地代表业务模型。业务实体被数据库的实现细节所绑架，每个字段、每个关系都必须与数据库表结构一一对应，完全失去了作为独立业务概念存在的自由。
更深层次的问题包括：
- Domain Layer 被 SQLAlchemy 绑定：业务实体继承了 SQLAlchemy 的 Base ，无法独立于数据库存在
- 业务逻辑无法脱离数据库测试：编写单元测试时必须启动完整的数据库环境，大大降低了测试效率
- 切换数据库需要修改所有层：当从 PostgreSQL 迁移到 MongoDB 时，所有使用 ORM Model 的代码都需要重写
。。。
![一个多 AI 的圆桌会议项目 [V1.0]1](https://xiaohack.oss-cn-zhangjiakou.aliyuncs.com/typecho/images/2026/01/11/20260111192756_6963893cd0d63.png!mark)
![一个多 AI 的圆桌会议项目 [V1.0]2](https://xiaohack.oss-cn-zhangjiakou.aliyuncs.com/typecho/images/2026/01/11/20260111192759_6963893fcd3b5.png!mark)