klr-pattern/pydantic-resolve/skill

pydantic-resolve 三阶段开发模式

基于 pydantic-resolve 的渐进式开发方法论。项目在一个 src/ 目录下逐步演进，每个阶段在上一阶段基础上新增代码。

核心原则

• 需求确认是 Phase 0，必须反复与用户确认后才能进入 Phase 1（详见下方「Phase 0: 需求确认」）
• 非功能模块与业务模块解耦，业务概念不侵入基础设施层
• 每个 Phase 采用 V 型验收：先定义验收标准（V 降），再实现，最后回查验收（V 升）
• 每个 Phase 实现完成后必须暂停，展示验收结果，等用户确认后再进入下一阶段
• Phase 间递进：同一项目目录下逐步丰富，只新增不修改已有代码

V 型验收模型（贯穿所有 Phase）

每个 Phase 的结构统一为三段：

┌──────────────────────────────────────────────┐
│ V 降：定义验收标准                              │
│   "在当前 Phase 开始之前，先定义什么算做完。"      │
│   写入 spec/<phase>.md 的"验收标准"部分            │
└──────────────────────────────────────────────┘
                      ↓
              ┌───────────────┐
              │   实现 Phase   │
              └───────────────┘
                      ↓
┌──────────────────────────────────────────────┐
│ V 升：逐条回查验收                             │
│   "一条一条对照验收标准，通过才可继续。"           │
│   用户逐条确认 → 写入 spec/<phase>.md             │
└──────────────────────────────────────────────┘

验收标准必须是可观察、可操作的——不写"代码健壮"，写"GraphiQL 中执行 X query 返回 Y"。

Phase 0: 需求确认（必做）

在写任何代码之前，必须与用户逐项确认以下内容。每一项都需要用户明确认可后才算完成。

Step 0-1: 术语与实体定义

逐一列出所有业务实体，每个实体说明：

• 业务含义（一句话，团队无歧义）
• 核心字段（名称 + 类型 + 语义说明，不需要穷举，但关键属性不能遗漏）
• 字段约束（唯一、非空、枚举值、联合唯一等）

用表格形式呈现，方便用户逐行确认。

Step 0-2: 实体关系

用文本 ER 图展示实体间关系，每条关系标明：

• 方向（1:N / N:1 / M:N）
• 业务含义（如「Sprint 包含多条 Task」）
• 是否需要中间实体

User ──1:N──→ Task
Sprint ──1:N──→ Task
Task ──N:1──→ User (owner)

必须与用户确认关系方向和基数是否正确。

Step 0-3: 聚合根

明确哪个（或哪些）实体是聚合根。聚合根决定：

• 主要的业务入口（从哪个实体开始查询）
• @query / @mutation 挂在哪些实体上
• Phase 3 的 service 划分依据

Step 0-4: 业务域划分 + 用例方法

⚠️ 禁止自行决定 Service 切分方案。必须提出候选方案与用户讨论，由用户最终确认。

Step 0-4a: 提出 Service 切分候选方案

业务域（Service）按功能边界划分，不按实体划分。Service 切分直接影响：

• 目录结构（service/<domain>/）
• Phase 2 的 methods.py 粒度
• Phase 3 的 UseCaseService 类划分
• MCP 和 REST 的入口组织

必须向用户提出至少一种候选方案，说明每种方案的切分依据和优劣，由用户选择或修正。

常见的切分策略参考：

Step 0-4b: 按确认的 Service 划分列出用例方法

用户确认 Service 切分后，按每个业务域列出用例方法。每个方法说明：

• 方法名（动词开头，如 list_sprints、create_task）
• 业务意图（一句话）
• 关键参数（列出参数名和含义）

示例格式：

Step 0-5: GraphQL 定位

GraphQL 是辅助开发测试和 AI 测试的接口，不是正式 API。

业务方法的定义和挂载关系：

service/<domain>/methods.py  ← 独立定义业务逻辑（核心）
        ↓ 挂载
  Entity via QueryConfig/MutationConfig
  (GraphQL 辅助测试)
 
  Phase 3: 同一个方法在 UseCaseService 中调用（REST/MCP 使用）

• Phase 2：方法体在 service/<domain>/methods.py 中实现，通过 QueryConfig/MutationConfig 绑定到 ErDiagram 的 Entity
• Phase 3：同一个方法在 UseCaseService 中调用（REST/MCP 使用），DTO 转换在 Service 层完成

Step 0-6: 第三方库确认

列出项目中涉及的非业务功能领域，对每个领域：

• 说明候选方案
• 给出推荐理由
• 必须调查用户提到的第三方库的当前维护状态

对于 pydantic-resolve 已覆盖的领域（ORM 集成、GraphQL、MCP），不再重复讨论。

Step 0-7: 检查清单

全部确认后，向用户展示汇总：

• [ ] 所有实体和字段是否完整，约束是否清晰？
• [ ] 实体关系方向和基数是否正确？
• [ ] 聚合根是否明确？
• [ ] Service 切分方案是否由用户确认？
• [ ] 核心用例是否覆盖主要业务场景？
• [ ] 第三方库选型是否确认？
• [ ] 是否有明显的遗漏或边界情况？

全部确认后才能进入 Phase 1。

参考实现

读取本 skill 目录下 template/ 中的代码作为生成参考。严格遵守 template 中的文件结构、import 风格和命名约定。

项目结构

单项目渐进演进，每个 Phase 在上一阶段基础上新增文件：

src/
├── db.py           # Phase 1（engine + session_factory，不依赖 models）
├── models.py       # Phase 1 SQLAlchemy ORM models（纯字段 + relationship）
├── entities.py     # Phase 1 Pydantic Entity DTOs + ErDiagram → Phase 2 新增 QueryConfig
├── database.py     # Phase 1（mock seed，依赖 db + models）
├── main.py         # 逐步扩展（voyager → graphql → rest → mcp）
├── service/        # Phase 2 新增 methods.py，Phase 3 补充 service.py/dtos.py
│   ├── sprint/
│   │   ├── methods.py  # Phase 2: 独立业务方法
│   │   ├── dtos.py     # Phase 3: DefineSubset DTOs
│   │   ├── service.py  # Phase 3: UseCaseService
│   │   └── spec.md     # Phase 3: 服务说明
│   ├── task/
│   │   ├── methods.py
│   │   ├── dtos.py
│   │   ├── service.py
│   │   └── spec.md
│   └── user/
│       └── methods.py
├── router/
│   └── api.py      # Phase 3: 手写 REST 路由

三阶段定义

Phase 1: Schema + ER Diagram + mock seed

目标: 定义 ORM models + Pydantic Entity DTOs + build_relationship() + mock seed data，用 Voyager 可视化供团队讨论。不含任何业务方法。

新增/修改文件:

• db.py — aiosqlite engine + session_factory（不导入 models，避免循环依赖）
• models.py — SQLAlchemy ORM models（纯字段 + relationship，加 lazy="noload"）
• entities.py — Pydantic Entity DTOs (from_attributes=True) + build_relationship() + ErDiagram + config_resolver()
• database.py — mock seed data（从 db.py 导入 engine/session，从 models.py 导入实体）
• main.py — FastAPI + Voyager（ER diagram 可视化）

关键模式:

• ORM Model 和 Entity DTO 分离：ORM Model 是 SQLAlchemy DeclarativeBase 子类，Entity DTO 是 Pydantic BaseModel 子类（加 model_config = ConfigDict(from_attributes=True)）
• build_relationship(mappings=[Mapping(entity=..., orm=...)], session_factory=...) 从 ORM 元数据自动生成所有 DataLoader
• ErDiagram(entities=[]).add_relationship(entities) 创建空 ErDiagram 后追加自动生成的关系
• config_resolver("MyResolver", er_diagram=diagram) 创建与 ErDiagram 绑定的 Resolver class
• ORM Model 的 relationship 全部加 lazy="noload"，不依赖 ORM 懒加载
• Voyager 使用 fastapi-voyager 的 create_voyager(app, er_diagram=diagram)
• Phase 1 无 GraphiQL（无方法可查询），GraphQL 在 Phase 2 后可用

# db.py
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
engine = create_async_engine("sqlite+aiosqlite://", echo=False)
async_session_factory = async_sessionmaker(engine, expire_on_commit=False)
def session_factory():
    return async_session_factory()

# models.py — 纯 ORM
from sqlalchemy import ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
class Base(DeclarativeBase):
    pass
class UserOrm(Base):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    tasks: Mapped[list["TaskOrm"]] = relationship(back_populates="owner", lazy="noload")

# entities.py — Pydantic DTOs + ErDiagram
from pydantic import BaseModel, ConfigDict
from pydantic_resolve import ErDiagram, config_resolver
from pydantic_resolve.integration.mapping import Mapping
from pydantic_resolve.integration.sqlalchemy import build_relationship
from src.db import session_factory
from src.models import UserOrm, TaskOrm, SprintOrm
class UserEntity(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str
# ... other entities ...
auto_entities = build_relationship(
    mappings=[
        Mapping(entity=UserEntity, orm=UserOrm),
        Mapping(entity=TaskEntity, orm=TaskOrm),
        Mapping(entity=SprintEntity, orm=SprintOrm),
    ],
    session_factory=session_factory,
)
diagram = ErDiagram(entities=[]).add_relationship(auto_entities)
MyResolver = config_resolver("MyResolver", er_diagram=diagram)

V 降 — 定义验收标准: 进入 Phase 1 实现之前，在 spec/phase1.md 中记录验收标准：

V 升 — 逐条回查验收:

• [ ] 1. Voyager ER 图：实体节点、关系线
• [ ] 2. ORM Model 纯字段：无业务方法
• [ ] 3. Entity DTO from_attributes=True
• [ ] 4. mock seed：数据量合理、关联关系正确

Phase 2: 方法实现 + Entity 挂载

目标: 按业务域实现独立方法，通过 QueryConfig/MutationConfig 绑定到 Entity，GraphQL 可查询。

新增/修改文件:

• service/<domain>/methods.py — 独立业务方法实现（直接操作 session + ORM model）
• entities.py — 新增 QueryConfig/MutationConfig 绑定

关键模式:

• 业务方法在 `service/<domain>/methods.py` 中定义，为普通 async 函数（直接操作 SQLAlchemy session + ORM model，不含 cls 参数）
• 通过 `QueryConfig`/`MutationConfig` 在 ErDiagram 层面绑定到 Entity（不需要 mount_method() + _mount() 桥接模式）
• QueryConfig(method=list_sprints, name="sprints") 中 method 是独立函数，name 是 GraphQL 查询名
• GraphQLHandler 接收 `er_diagram=diagram`（不是 base + session_factory）
• config_resolver 必须在 build_relationship 之后调用
• GraphQLHandler 必须在 ErDiagram 完整配置（含 QueryConfig/MutationConfig）之后创建

# service/sprint/methods.py
from sqlalchemy import select
from src.db import session_factory
from src.models import SprintOrm
async def list_sprints() -> list[SprintOrm]:
    """获取所有 Sprint。"""
    async with session_factory() as session:
        result = await session.execute(select(SprintOrm).order_by(SprintOrm.id))
        return list(result.scalars().all())

# entities.py — 新增 QueryConfig/MutationConfig（Phase 2 新增部分）
from pydantic_resolve import Entity, QueryConfig, MutationConfig
from src.service.sprint.methods import list_sprints, get_sprint, create_sprint
query_mutation_entities = [
    Entity(kls=SprintEntity, queries=[
        QueryConfig(method=list_sprints, name="sprints"),
        QueryConfig(method=get_sprint, name="sprint"),
    ], mutations=[
        MutationConfig(method=create_sprint, name="create_sprint"),
    ]),
    # ... other entities ...
]
diagram = (
    ErDiagram(entities=[])
    .add_relationship(auto_entities)
    .add_relationship(query_mutation_entities)
)
MyResolver = config_resolver("MyResolver", er_diagram=diagram)

V 降 — 定义验收标准: 进入 Phase 2 编码之前，先与用户确认测试验收集并写入 spec/phase2.md：

验收标准要求：

• 每个 query/mutation 至少覆盖：一个正常场景
• 验证方式统一通过 GraphQL query/mutation 在 GraphiQL 中执行

V 升 — 逐条回查验收: 启动服务，在 GraphiQL 中逐一执行验收表：

• [ ] 1. list_sprints → 返回 seed 数据
• [ ] 2. create_sprint → 返回新 Sprint 对象
• [ ] 3. tasks_by_sprint → 返回过滤后的任务列表
• [ ] ...（每条对标验收表）
• [ ] 确认 seed 数据仍可查询，Loader 行为符合预期

Phase 3: UseCase 响应组装 + MCP + REST

目标: 按 API 用例组装响应结构。DefineSubset 隐藏内部字段，AutoLoad 自动加载关系，UseCaseService 统一业务入口，REST 端点对外暴露。

新增/修改文件:

• service/<domain>/spec.md — 服务目的、用途、需求、变更记录
• service/<domain>/dtos.py — DefineSubset DTOs + AutoLoad + post_*
• service/<domain>/service.py — UseCaseService
• router/api.py — 手写 REST 路由
• main.py — 新增 REST + MCP