بنقرة واحدة
pydantic-resolve-3step
// 基于 pydantic-resolve 的三阶段开发模式,用于构建从建模讨论到生产部署的完整项目。适用于需要使用 ER Diagram + ORM + DefineSubset 渐进式开发的场景。
// 基于 pydantic-resolve 的三阶段开发模式,用于构建从建模讨论到生产部署的完整项目。适用于需要使用 ER Diagram + ORM + DefineSubset 渐进式开发的场景。
| name | pydantic-resolve-3step |
| description | 基于 pydantic-resolve 的三阶段开发模式,用于构建从建模讨论到生产部署的完整项目。适用于需要使用 ER Diagram + ORM + DefineSubset 渐进式开发的场景。 |
| argument-hint | [项目路径] 创建三阶段项目的目标目录 |
基于 pydantic-resolve 的渐进式开发方法论。项目在一个 src/ 目录下逐步演进,每个阶段在上一阶段基础上新增代码。
| Phase | 职责 | 产出 |
|---|---|---|
| Phase 0 | 需求确认 | 实体 + 关系 + 聚合根 + 用例方法(与用户反复确认) |
| Phase 1 | Schema + ER Diagram + mock seed | ORM models + Entity DTOs + build_relationship + Voyager |
| Phase 2 | 方法实现 + GraphQL | service//methods.py + QueryConfig/MutationConfig + GraphQL |
| Phase 3 | UseCase 响应组装 + MCP + REST | DefineSubset + AutoLoad + UseCaseService + REST + MCP |
每个 Phase 的结构统一为三段:
┌──────────────────────────────────────────────┐
│ V 降:定义验收标准 │
│ "在当前 Phase 开始之前,先定义什么算做完。" │
│ 写入 spec/<phase>.md 的"验收标准"部分 │
└──────────────────────────────────────────────┘
↓
┌───────────────┐
│ 实现 Phase │
└───────────────┘
↓
┌──────────────────────────────────────────────┐
│ V 升:逐条回查验收 │
│ "一条一条对照验收标准,通过才可继续。" │
│ 用户逐条确认 → 写入 spec/<phase>.md │
└──────────────────────────────────────────────┘
验收标准必须是可观察、可操作的——不写"代码健壮",写"GraphiQL 中执行 X query 返回 Y"。
在写任何代码之前,必须与用户逐项确认以下内容。每一项都需要用户明确认可后才算完成。
逐一列出所有业务实体,每个实体说明:
用表格形式呈现,方便用户逐行确认。
用文本 ER 图展示实体间关系,每条关系标明:
User ──1:N──→ Task
Sprint ──1:N──→ Task
Task ──N:1──→ User (owner)
必须与用户确认关系方向和基数是否正确。
明确哪个(或哪些)实体是聚合根。聚合根决定:
⚠️ 禁止自行决定 Service 切分方案。必须提出候选方案与用户讨论,由用户最终确认。
业务域(Service)按功能边界划分,不按实体划分。Service 切分直接影响:
service/<domain>/)必须向用户提出至少一种候选方案,说明每种方案的切分依据和优劣,由用户选择或修正。
常见的切分策略参考:
| 策略 | 示例 | 适用场景 |
|---|---|---|
| 按业务功能域 | auth / chat / order | 业务边界清晰,领域间耦合低 |
| 按聚合根 | user / conversation / message | 实体独立性强,CRUD 为主 |
| 混合(功能域 + 独立聚合) | auth / chat(含 conversation+message) | 部分域跨实体协作 |
用户确认 Service 切分后,按每个业务域列出用例方法。每个方法说明:
list_sprints、create_task)示例格式:
| 业务域 | 方法名 | 业务意图 | 关键参数 |
|---|---|---|---|
| sprint | list_sprints | 获取所有 Sprint | - |
| sprint | get_sprint | 获取单个 Sprint | sprint_id |
| task | list_tasks | 获取所有任务 | - |
| task | get_tasks_by_sprint | 按 Sprint 查任务 | sprint_id |
| task | create_task | 创建任务 | title, sprint_id, owner_id |
GraphQL 是辅助开发测试和 AI 测试的接口,不是正式 API。
业务方法的定义和挂载关系:
service/<domain>/methods.py ← 独立定义业务逻辑(核心)
↓ 挂载
Entity via QueryConfig/MutationConfig
(GraphQL 辅助测试)
Phase 3: 同一个方法在 UseCaseService 中调用(REST/MCP 使用)
service/<domain>/methods.py 中实现,通过 QueryConfig/MutationConfig 绑定到 ErDiagram 的 Entity列出项目中涉及的非业务功能领域,对每个领域:
| 功能领域 | 推荐方案 | 理由 | 备注 |
|---|---|---|---|
| ORM | SQLAlchemy 2.0 async | 成熟、pydantic-resolve 原生支持 build_relationship | - |
| 可视化 | fastapi-voyager | ER diagram 可视化 | pip install fastapi-voyager |
| MCP | fastmcp>=3.2.4 | pydantic-resolve 原生支持 | - |
对于 pydantic-resolve 已覆盖的领域(ORM 集成、GraphQL、MCP),不再重复讨论。
全部确认后,向用户展示汇总:
全部确认后才能进入 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 路由
目标: 定义 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 可视化)关键模式:
DeclarativeBase 子类,Entity DTO 是 Pydantic BaseModel 子类(加 model_config = ConfigDict(from_attributes=True))build_relationship(mappings=[Mapping(entity=..., orm=...)], session_factory=...) 从 ORM 元数据自动生成所有 DataLoaderErDiagram(entities=[]).add_relationship(entities) 创建空 ErDiagram 后追加自动生成的关系config_resolver("MyResolver", er_diagram=diagram) 创建与 ErDiagram 绑定的 Resolver classlazy="noload",不依赖 ORM 懒加载fastapi-voyager 的 create_voyager(app, er_diagram=diagram)# 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 中记录验收标准:
| # | 验收项 | 验证方式 |
|---|---|---|
| 1 | 每个 Entity 在 Voyager ER 图中正确显示,关系线方向正确 | 浏览器打开 Voyager |
| 2 | models.py 中每个 ORM Model 只包含字段 + relationship,无业务方法 | 检查代码结构 |
| 3 | entities.py 中每个 Entity DTO 有 from_attributes=True | 检查代码 |
| 4 | mock seed 数据样本展示合理的数量、关联关系 | 编写简单查询验证 |
V 升 — 逐条回查验收:
目标: 按业务域实现独立方法,通过 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 查询名er_diagram=diagram(不是 base + session_factory)# 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:
| # | 方法 | 测试场景 | 预期结果 | 验证方式 |
|---|---|---|---|---|
| 1 | list_sprints | 查询所有 | 返回 seed 数据 | GraphiQL query |
| 2 | create_sprint | 创建新 Sprint | 返回新 Sprint | GraphiQL mutation |
| 3 | tasks_by_sprint | 按 sprint 查任务 | 返回正确过滤结果 | GraphiQL query |
| ... | ... | ... | ... | ... |
验收标准要求:
V 升 — 逐条回查验收: 启动服务,在 GraphiQL 中逐一执行验收表:
目标: 按 API 用例组装响应结构。DefineSubset 隐藏内部字段,AutoLoad 自动加载关系,UseCaseService 统一业务入口,REST 端点对外暴露。
新增/修改文件:
service/<domain>/spec.md — 服务目的、用途、需求、变更记录service/<domain>/dtos.py — DefineSubset DTOs + AutoLoad + post_*service/<domain>/service.py — UseCaseServicerouter/api.py — 手写 REST 路由main.py — 新增 REST + MCP关键模式:
__subset__ = (Entity, ['id', 'name'])origin 参数映射关系名:Annotated[UserSummary | None, AutoLoad(origin="owner")] = Nonelist[OrmModel] → [Dto.model_validate(m) for m in models] → MyResolver(enable_from_attribute_in_type_adapter=True).resolve(dtos)OrmModel | None → Dto.model_validate(entity) → (await MyResolver(...).resolve([dto]))[0]create_use_case_router())enable_from_attribute_in_type_adapter=True 在从 ORM 转换时必须设置create_use_case_mcp_server() + UseCaseAppConfigtransport="streamable-http", stateless_http=True# service/task/dtos.py
from typing import Annotated
from pydantic_resolve import DefineSubset, AutoLoad
from src.entities import UserEntity, TaskEntity
class UserSummary(DefineSubset):
__subset__ = (UserEntity, ["id", "name"])
class TaskSummary(DefineSubset):
__subset__ = (TaskEntity, ["id", "title", "done"])
owner_detail: Annotated[UserSummary | None, AutoLoad(origin="owner")] = None
# service/sprint/dtos.py
from pydantic_resolve import DefineSubset, AutoLoad
from src.entities import SprintEntity, TaskEntity
from src.service.task.dtos import TaskSummary
class SprintSummary(DefineSubset):
__subset__ = (SprintEntity, ["id", "name"])
task_list: Annotated[list[TaskEntity], AutoLoad(origin="tasks")] = []
task_count: int = 0
contributor_names: list[str] = []
def post_task_count(self):
return len(self.task_list)
def post_contributor_names(self):
return sorted({t.owner.name for t in self.task_list if t.owner})
# service/task/service.py
from pydantic_resolve import query
from pydantic_resolve.use_case import UseCaseService
from src.entities import MyResolver
from src.service.task.dtos import TaskSummary
from src.service.task.methods import list_tasks as _list_tasks
class TaskService(UseCaseService):
@query
async def list_tasks(cls) -> list[TaskSummary]:
tasks = await _list_tasks()
dtos = [TaskSummary.model_validate(t) for t in tasks]
return await MyResolver(enable_from_attribute_in_type_adapter=True).resolve(dtos)
# router/api.py
from fastapi import APIRouter, HTTPException
from src.service.sprint.service import SprintService
from src.service.task.service import TaskService
route = APIRouter(prefix="/api")
@route.get("/tasks", tags=[TaskService.get_tag_name()])
async def get_tasks():
return await TaskService.list_tasks()
@route.get("/sprints", tags=[SprintService.get_tag_name()])
async def get_sprints():
return await SprintService.list_sprints()
V 降 — 定义验收标准:
| # | 验收项 | 验证方式 |
|---|---|---|
| 1 | REST 端点返回的响应字段符合 DTO 定义(FK 隐藏、关系字段包含) | curl GET endpoint |
| 2 | MCP 四层渐进式披露完整:discover → inspect → execute | MCP 客户端调用 |
| 3 | GraphQL 查询仍然正常工作 | GraphiQL |
V 升 — 逐条回查验收:
curl /api/tasks 返回字段符合 TaskSummary DTO,不含 sprint_id/owner_idcurl /api/sprints 返回 SprintSummary 含 task_count 和 contributor_names| 方面 | Phase 1 | Phase 2 | Phase 3 |
|---|---|---|---|
| ORM Model | 纯字段 + relationship (noload) | 不变 | 不变 |
| Entity DTO | from_attributes=True | 新增 QueryConfig/MutationConfig | 继承 |
| 关系 | build_relationship() 自动生成 | 不变 | DefineSubset 隐藏 FK |
| 查询函数 | 无方法 | methods.py + QueryConfig | UseCaseService 封装(复用 methods.py) |
| API | Voyager(ER diagram) | GraphQL | GraphQL + REST + MCP |
| 响应 | N/A | 完整实体 | DefineSubset DTO |
db.py — models.py 和 database.py 都需要 session,放在同一文件会导致循环导入pyproject.toml 必须配置 packages = ["src"] — hatchling 需要 [tool.hatch.build.targets.wheel]from __future__ import annotations — 会使类型注解变字符串,SubsetMeta 无法检测 Annotated 元数据,FK 字段不会被自动添加from_attributes=True 会尝试访问所有 ORM 属性(包括未加载的 relationship),导致 DetachedInstanceErrorbuild_relationship 需要 Entity 有 from_attributes=True — Mapping 验证 Entity 的 model_config 必须包含 from_attributes=Trueconfig_resolver 必须在 build_relationship 之后调用 — resolver 需要完整的 ErDiagramGraphQLHandler 必须在 ErDiagram 完整配置后创建 — handler 初始化时扫描 Entity 的 query/mutation 方法resolve([dto]) 后取 [0] — Resolver 只接受 listenable_from_attribute_in_type_adapter=True 在从 ORM 转换时需要 — 否则 DataLoader 返回的 ORM 对象无法被 TypeAdapter 正确处理lazy="noload" — 不依赖 ORM lazy-load,避免 session 关闭后 DetachedInstanceErrortransport="streamable-http", stateless_http=True,在 FastAPI lifespan 中嵌套 MCP http_app 的 lifespanQueryConfig 的 name 参数不是最终 GraphQL 查询名 — 实际名称格式为 {EntityName}{NameCamel},如 QueryConfig(method=list_sprints, name="sprints") 对应 sprintEntitySprintsAutoLoad(origin="relationship_name") 映射GraphQLHandler.execute() 只接受 query 和 context — 不接受 variables 和 operation_name,与 sqlmodel-nexus 不同每次使用 skill 时,必须在项目根目录下创建 spec/ 目录:
spec/<编号>-<需求简述>/
YY-MM-DD + 两位序号,如 250510-01示例: spec/250510-01-sprint-demo/
spec/<编号>-<需求简述>/
├── story.md # 用户原始需求 + Overview Design
├── phase0.md # 需求确认
├── phase1.md # Schema + ER Diagram
├── phase2.md # 方法实现 + GraphQL
└── phase3.md # UseCase + MCP + REST
每个 phase 文件分三个部分:
# Phase N: <阶段标题>
## 需求说明
(记录用户在对话中提出的原始需求、约束条件和确认结论)
## 验收标准
(V 降阶段定义的验收标准表格)
## 实现描述
(记录技术实现方案、产出文件、关键决策和 V 升回查结果)
| 文件 | 写入时机 |
|---|---|
| story.md | 用户首次描述需求时记录原始表述;Phase 0 确认后补充 Overview Design |
| phase0.md | Phase 0 全部确认后 |
| phase1.md | V 降 → 实现 → V 升通过后 |
| phase2.md | V 降 → 实现 → V 升通过后 |
| phase3.md | V 降 → 实现 → V 升通过后 |
当用户要求创建三阶段项目时:
spec/<编号>-<需求简述>/,将原始需求写入 story.mdphase0.md → 补充 story.md 的 Overview Designspec/phase1.md#验收标准phase1.md → 暂停等用户确认spec/phase2.md#验收标准phase2.md → 暂停等用户确认spec/phase3.md#验收标准phase3.md → 暂停等用户确认Phase 0 全部确认后、进入 Phase 1 之前,在 story.md 中补充 ## Overview Design: