python

《Python编程实战》18.1 需求拆解与架构设计

全书收尾章从一个贯穿三节的真实项目 TaskFlow 出发：把模糊需求拆成用户故事与用例、画出领域模型与状态机、用 OpenAPI 固化 API 契约，再落到技术选型表、架构图、目录结构、pyproject.toml，并用 Alembic 真跑通第一版数据库迁移。

Leeting Yan 7 分钟阅读 3505 字

本节目标：拿到一句模糊的需求后，能把它拆成用户故事、用例、领域模型与可执行的 API 契约，并落成目录结构、pyproject.toml 与第一版数据库迁移。
适用版本：Python 3.12+（实测 3.14.6）；SQLAlchemy 2.1.4、alembic 1.20.0

18.1 需求拆解与架构设计

前面 17 章把工具箱摆满了：脚手架、依赖、配置、日志、测试、FastAPI、SQLAlchemy、Pydantic、缓存、任务队列、认证、部署、剖析、可观测性。但真实项目不是「一章一个技术点」，而是一堆需求压着你，要在有限时间里做出能上线的系统。这一章用一个贯穿三节的项目把它们串起来。

项目叫 TaskFlow：一个任务/工单管理系统的 REST API，带用户认证、SQLite/PostgreSQL 数据层、后台任务与缓存。它足够小，能在这三节里真的写完并跑起来；又足够完整，覆盖了前面每一章的关键决策。

18.1.1 从一句话需求到用户故事

需求原文只有一句：「我们要一个内部工单系统，员工能提单、能看进度、能留言。」

这种句子无法直接开工——它没定义角色、边界、验收标准。第一步是拆成用户故事（User Story），每条都写成「作为〈角色〉，我想要〈能力〉，以便〈价值〉」：

#用户故事优先级
US-1作为员工，我想注册并登录，以便系统知道我是谁P0
US-2作为员工，我想提交工单（标题、描述、优先级），以便问题被记录P0
US-3作为员工，我想查看自己的工单列表并分页，以便追踪进度P0
US-4作为员工，我想更新工单状态，以便反映处理进展P0
US-5作为员工，我想给工单留言，以便补充信息P1
US-6作为员工，我想只看到自己的工单，以便数据隔离P0

注意 US-6：需求原文根本没提，但「内部系统」隐含了多租户/数据隔离——这是必须在设计阶段就定下的非功能需求，否则后期补隔离会推倒重来。P0 的六条正好构成一个最小可用产品（MVP）。

18.1.2 用例：把故事拆成可验证的步骤

用户故事描述「要什么」，用例（Use Case）描述「怎么做、成功/失败长什么样」。以 US-2「提交工单」为例：

用例 UC-2：提交工单
  参与者：已登录员工
  前置条件：持有有效访问令牌
  主流程：
    1. 客户端 POST /tickets，带标题、描述、优先级
    2. 服务端校验令牌 -> 解析出当前用户
    3. 校验请求体（标题非空、优先级合法）
    4. 落库，状态初始为 open，优先级默认 medium
    5. 投递一条 notify 后台任务
    6. 返回 201 与工单对象
  异常流程：
    a. 无令牌/令牌失效 -> 401
    b. 标题为空 -> 422（由 Pydantic 拦截）

用例的价值在于：每一步都能变成一条测试。主流程第 6 步对应「创建返回 201」，异常 a 对应「未认证访问返回 401」——这些正是 18.2 里 pytest 要覆盖的。

18.1.3 领域模型：三个实体与一条状态机

从用例里能提炼出三个实体：User（用户）、Ticket（工单）、Comment（评论）。关系是：一个用户有多张工单，一张工单有多条评论。

  User 1 ────< Ticket 1 ────< Comment
   │                              │
   └──────────────<───────────────┘
        （Comment 同时指向 User 作为作者）

工单状态必须定义成有限状态机，否则 status 会退化成谁都能乱写的字符串：

  open ──→ in_progress ──→ resolved ──→ closed
    └──────────────────────────┘
       （允许直接关闭，禁止从 closed 回退）
字段类型约束说明
titlestr1–200 字符必填
bodystr0–4000 字符可空
statusenumopen/in_progress/resolved/closed默认 open
priorityenumlow/medium/high/urgent默认 medium
owner_idint外键 → users.id，建索引隔离依据

18.1.4 API 契约先行：OpenAPI 就是可执行的契约

需求一旦拆到用例层，就能直接写成接口清单。契约先行（Contract-First） 的意思是：先把接口的形状定死，前后端并行开工。下面是本项目的真实契约（由 FastAPI 自动生成的 app.openapi() 导出）：

方法路径说明认证
POST/auth/register注册否
POST/auth/login登录换令牌否
GET/healthz健康探针否
POST/tickets创建工单是
GET/tickets分页列出自己的工单是
GET/tickets/{ticket_id}查单张工单是
PATCH/tickets/{ticket_id}改状态/优先级是
DELETE/tickets/{ticket_id}删除工单是
POST/tickets/{ticket_id}/comments加评论是

契约先行的三个收益：并行开发（前端照着 schema 造 mock）、自动文档（/docs 免费获得）、可测（每个端点都对应一条集成测试）。第 7 章讲过怎么用 OpenAPI 生成客户端，这里不再展开。

18.1.5 技术选型：每个选择都要能说出为什么

选型不是「哪个火用哪个」，而是每条需求对应一个决策。把决策写进表格，评审时才有得聊：

需求选型理由备选
HTTP 服务FastAPI 0.143.0异步、自带校验与 OpenAPIFlask（无异步）
数据层SQLAlchemy 2.1.4 async类型化模型、异步、多方言裸 sqlite3（无 ORM）
本地库SQLite + aiosqlite 0.22.1零运维，生产可切 PostgreSQLPostgreSQL（本机无服务端）
校验Pydantic 2.13.5与 FastAPI 同源，边界校验手写校验
配置pydantic-settings 2.15.0环境变量分层，类型安全os.environ
认证HS256 JWT（自实现）无额外依赖，教学透明PyJWT（本机未装）
缓存fakeredis 2.39.0内存实现，API 兼容 Redis真实 Redis（本机无）
迁移alembic 1.20.0版本化 schema 演进手工 SQL
测试pytest 9.1.1 + httpx 0.28.1ASGI 内联测试，快起真服务

⚠️ 诚实标注：本机没有 PostgreSQL 服务端、没有 Redis 服务端、没有 PyJWT。所以数据层用 SQLite 实测（SQLAlchemy 的 PG 专属方言未实测），缓存用 fakeredis 实测，JWT 用标准库 hmac 自实现。生产切到 PostgreSQL + 真实 Redis 时 API 不变，但连接串与方言行为需重新验证。

18.1.6 架构图与目录结构

选型定完，画出数据流。这张图三节都会用到：

         ┌────────────┐  HTTPS   ┌───────────────────────────────┐
客户端 → │ httpx/curl │ ───────→ │ uvicorn (ASGI server)         │
         └────────────┘          │  └─ FastAPI app               │
                                 │     ├─ 中间件：计时/请求日志   │
                                 │     ├─ /auth    认证          │
                                 │     ├─ /tickets 工单 + 评论   │
                                 │     └─ /healthz 探针          │
                                 └──────┬────────────────┬───────┘
                                        │                │
                                 ┌──────▼──────┐  ┌──────▼──────┐
                                 │ SQLAlchemy  │  │  fakeredis  │
                                 │ 2.1.4 async │  │ 缓存 + 队列 │
                                 └──────┬──────┘  └─────────────┘
                                        │
                                 ┌──────▼──────┐
                                 │ SQLite /    │
                                 │ PostgreSQL  │
                                 └─────────────┘

目录结构按「分层 + 按功能切分路由」组织，这也是第 5 章推荐的形状：

taskflow/
├── app/
│   ├── config.py          # pydantic-settings 配置
│   ├── db.py              # 引擎、会话工厂、Base
│   ├── models.py          # SQLAlchemy 类型化模型
│   ├── schemas.py         # Pydantic 请求/响应模型
│   ├── security.py        # 密码哈希 + JWT
│   ├── cache.py           # fakeredis 缓存封装
│   ├── tasks.py           # 后台任务队列（幂等/重试/死信）
│   ├── deps.py            # 依赖注入（会话、当前用户）
│   ├── main.py            # app 装配、lifespan、中间件
│   └── routers/
│       ├── auth.py
│       └── tickets.py
├── tests/                 # conftest + 单元/集成测试
├── bench/                 # 端到端联调 + 压测脚本
├── alembic/               # 数据库迁移
└── pyproject.toml

依赖声明写在 pyproject.toml 里（第 1、2 章的工程化起点）：

[project]
name = "taskflow"
version = "1.0.0"
requires-python = ">=3.12"
dependencies = [
    "fastapi==0.143.0",
    "uvicorn==0.54.0",
    "sqlalchemy==2.1.4",
    "aiosqlite==0.22.1",
    "pydantic==2.13.5",
    "pydantic-settings==2.15.0",
    "fakeredis==2.39.0",
    "httpx==0.28.1",
]

[tool.pytest.ini_options]
asyncio_mode = "auto"
pythonpath = ["."]
testpaths = ["tests"]

所有版本号都精确固定——第 2 章讲过，== 锁死版本是可复现构建的前提。

18.1.7 数据库 schema 与迁移起步

模型定完，第一件事不是手写建表 SQL，而是让 Alembic 从模型自动生成迁移。初始化（真实执行）：

alembic init -t async alembic     # -t async 生成异步版 env.py

改两处 env.py：把 target_metadata 指向模型元数据，并让 URL 走配置：

from app.config import settings
from app.db import Base
from app import models  # noqa: F401  确保模型被导入，autogenerate 才看得到

target_metadata = Base.metadata
# run_async_migrations() 里：
config.set_main_option("sqlalchemy.url", settings.database_url)

然后自动生成并应用首个迁移（真实输出）：

$ alembic revision --autogenerate -m "init tickets schema"
INFO  [alembic.autogenerate.compare.tables] Detected added table 'users'
INFO  [alembic.autogenerate.compare.tables] Detected added table 'tickets'
INFO  [alembic.autogenerate.compare.tables] Detected added table 'comments'
Generating .../versions/9ae51e968831_init_tickets_schema.py ...  done

$ alembic upgrade head
INFO  [alembic.runtime.migration] Running upgrade  -> 9ae51e968831, init tickets schema

迁移跑完，SQLite 里真实生成的 DDL 长这样：

CREATE TABLE tickets (
	id INTEGER NOT NULL,
	title VARCHAR(200) NOT NULL,
	body VARCHAR(4000) NOT NULL,
	status VARCHAR(11) NOT NULL,
	priority VARCHAR(6) NOT NULL,
	owner_id INTEGER NOT NULL,
	created_at DATETIME DEFAULT (CURRENT_TIMESTAMP) NOT NULL,
	PRIMARY KEY (id),
	FOREIGN KEY(owner_id) REFERENCES users (id)
);
CREATE INDEX ix_tickets_owner_id ON tickets (owner_id);

一个真实踩到的细节：status 列被自动生成为 sa.Enum('OPEN', 'IN_PROGRESS', ...)——SQLAlchemy 的 Enum 默认按枚举成员名（大写）落库，而不是成员值（小写）。API 层 Pydantic 序列化时用的是值 open，所以数据库里存 OPEN、接口返回 open。这个「名字 vs 值」的错位不影响功能，但排查数据时极易困惑，务必知道。

延伸阅读

小结

  • 一句需求不能直接开工：先拆用户故事（角色/能力/价值），再拆用例（主流程 + 异常流程），每一步都能变成测试。
  • 非功能需求（如数据隔离）必须在设计阶段定下，否则后期补会推倒重来。
  • 领域模型 + 状态机把 status 从自由字符串变成受约束的枚举。
  • 契约先行：先用 OpenAPI 定死接口形状，前后端并行、文档免费、测试有据。
  • 技术选型表强迫你为每个决策写出理由与备选，评审时才有得聊。
  • Alembic 从模型自动生成迁移，不手写建表 SQL；注意 Enum 默认按成员名落库这一真实陷阱。

需求拆完、架构画好、迁移跑通，地基就有了。下一节开始真的把它写出来并跑起来——路由、模型、测试、联调、压测，一行行落到可运行的代码。

阅读导航：上一节：灰度发布、回滚与故障演练 · 下一节：迭代开发、联调与压测 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「python」更多文章

  1. 《Python高级编程》目录
  2. 《Python高级编程》11.3 PEP 流程与版本迁移策略
  3. 《Python高级编程》11.2 嵌入式与自由线程运行时