一个课程报名接口，通常要干这几件事：读取学生、读取课程、判断有没有重复报名、看看课程满没满，最后把报名记录写进数据库。
在快速迭代时，我们往往会把这些步骤一股脑全塞进一个 HTTP 路由函数里。这确实能很快跑起来，但随着需求增加，业务规则、数据库 SQL 和 HTTP 状态码全都会死死地缠绕在一起（俗称“面条代码”）。以后无论是想加个规则、换个数据库，还是仅仅想写个独立的单元测试，都会变得异常痛苦。
本文以 Hucci写代码的《5 分钟学会写架构设计》视频及同主题笔记中的报名例子为线索，写出一个可运行的 Python 版本。我们将重点探讨：在优秀的架构中，每段代码究竟该负责什么？谁可以调用谁？
面条代码的痛点在哪里？
下面是常见的“一锅炖”写法示意（省略了建表与连接管理）：
1@app.post("/enrollments")
2def enroll(request):
3 student = db.execute("SELECT ... FROM students WHERE id = ?", ...)
4 course = db.execute("SELECT ... FROM courses WHERE id = ?", ...)
5 if student is None or course is None:
6 raise HTTPException(404, "不存在")
7 if db.execute("SELECT ... FROM enrollments WHERE ...", ...):
8 raise HTTPException(409, "重复报名")
9 db.execute("INSERT INTO enrollments ...", ...)
10 return {"ok": True}
这段代码的问题不在于行数，而在于它同时承担了三种可能发生的变化：
- 协议变了：如果你想把 HTTP 接口改成 gRPC 或者 Kafka 消息消费，这段代码就废了。
- 存储变了：如果你想把数据库从 SQLite 换成 MySQL 或 Redis，这段代码也得大改。
- 规则变了：哪怕只是加一个“VIP学生免排队”的规则，你也得在一堆 SQL 和 HTTP 状态码中小心翼翼地修改。
变化相互牵连，这就是代码越来越难改的根本原因。
用“开餐厅”的直觉理解四层架构
把所有逻辑写在一个函数里，就像一家餐厅里，服务员不仅要负责在前台点单，还要自己跑到后厨切菜、炒菜，最后还要去仓库盘点食材。一旦客人多了或换个菜系，餐厅直接崩溃。
因此，我们需要将系统分层。分层有两层核心原则：
- 职责单一：每层只干自己该干的事。
- 依赖方向：让容易变化的层，依赖不容易变化的层。
这第二条尤其重要，也是整篇文章想讲透的一句话。一段代码的变化频率是天生不同的：
api变化最频繁：HTTP 状态码、请求/响应格式、接口路径，三天两头就可能调整；service次之：报名流程的先后步骤偶尔会变；db的实现也可能变：今天 SQLite，明天可能换成 PostgreSQL；domain最稳定：学生、课程这些概念，以及“不能重复报名、不能超额”这些规则，是业务的地基，很少变动。
如果让稳定的层去依赖易变的层（比如 domain 里 import 了某个 HTTP 框架），那么 HTTP 一改，核心规则就得跟着改，系统就会被反复拖累。反过来，让易变的层（api）去依赖稳定的层（service、domain），不管外面怎么翻腾，内部地基都稳如泰山。
让容易变化的层，依赖不容易变化的层。 箭头永远指向更稳定的地方。
在这个 Python 项目中，我们将代码按职责拆分为四层，你可以这样直观地理解它们：
domain(领域层) —— “后厨的大厨与核心菜谱”- 作用：封装最核心的业务概念（学生、课程）和死磕业务规则（不存在、重复报名、满员）。
- 铁律：它是系统的绝对内层，纯纯的 Python 代码，不能有任何 SQL，不能有任何 HTTP 框架的导入。这使得它可以在不连数据库的情况下，1 毫秒内跑完几千个单元测试。
service(服务层) —— “大堂经理”- 作用：编排业务流程。他知道整个报名的先后步骤（查人 -> 查课 -> 校验规则 -> 存入数据库），但他自己不干苦力。他向
db拿数据，交给domain校验，再让db保存。 - 铁律：他不直接依赖具体的数据库类，而是声明自己需要一个“仓管员”（
EnrollmentStore接口）。这样，底层怎么换数据库，经理根本不关心。
- 作用：编排业务流程。他知道整个报名的先后步骤（查人 -> 查课 -> 校验规则 -> 存入数据库），但他自己不干苦力。他向
db(数据存取层) —— “仓库管理员”- 作用：负责与数据库进行真实的物理交互。写 SQL、建连接、做事务。
- 铁律：只管按
service的要求存取数据，绝不擅自做主去校验“业务规则”，也不操心 HTTP 怎么返回。
api(接口层 / 表现层) —— “前台接待”- 作用：站在最外面接客。接收外部 HTTP 请求，转交给
service去干活，然后把结果或错误翻译成 200、404 状态码返回给客户。同时，它是系统的装配点（它负责把具体的 DB 实例指派给 Service）。 - 铁律：路由函数里绝对不写任何核心业务判断和 SQL。
- 作用：站在最外面接客。接收外部 HTTP 请求，转交给
它们之间的调用链非常清晰：HTTP 请求 → api → service → domain。
箭头指向的地方就是被依赖的地方，也是更稳定、更不容易变化的地方。越往内层，代码越稳定、越纯粹——这正是“让容易变化的层，依赖不容易变化的层”的直观体现。
把依赖关系画出来就是这样：
flowchart LR
HTTP["HTTP 请求"] --> api["api<br/>前台接待"]
api --> service["service<br/>大堂经理"]
service --> domain["domain<br/>大厨/规则"]
service -. 依赖（声明所需存储能力） .-> store["EnrollmentStore<br/>存储接口 / 岗位说明书"]
db["db<br/>仓库管理员"] -. 实现 .-> store从图里能看出两件事：
- 实线箭头永远指向更稳定的内层：
api依赖service，service依赖domain，谁变化频繁谁站在外侧、去依赖里面更稳的。 service依赖的是“接口”而不是db本身：service只认EnrollmentStore这份“岗位说明书”，而db主动去实现它。箭头的方向，决定了换掉db（比如 SQLite 换 PostgreSQL）时，service一行都不用改。
注意：如何理解依赖倒置？
在运行时，大堂经理（
service）确实是指挥仓库管理员（具体的 SQLite 数据库对象）干活的。但在代码编写上，经理只看“岗位说明书”（自己定义的EnrollmentStore接口）。只要 SQLite 类通过 Python 的Protocol类型满足了这个岗位的要求，经理就能用它。这样一来，经理的代码里完全没有import sqlite3，这就是依赖倒置。
写一个能运行的报名例子
使用 Python 3.10+，创建下面的目录。代码也保存在本文所在目录的 src/registration_layers/ 中；下文每一段代码都可直接复制到对应文件。
1registration_demo/
2├── registration_layers/
3│ ├── __init__.py
4│ ├── domain.py
5│ ├── service.py
6│ ├── db.py
7│ └── api.py
8└── .venv/ # 安装依赖后生成
1. domain：把规则写成普通 Python
check_enrollment 只接收数据和当前状态。无需连接数据库，也无需构造 HTTP 请求，就能验证各种边界情况。
1"""Business concepts and rules; no HTTP or database imports."""
2
3from dataclasses import dataclass
4
5
6@dataclass(frozen=True)
7class Student:
8 id: int
9 name: str
10
11
12@dataclass(frozen=True)
13class Course:
14 id: int
15 title: str
16 capacity: int
17
18
19class EnrollmentError(Exception):
20 """Expected failure of an enrollment request."""
21
22
23class StudentNotFound(EnrollmentError):
24 pass
25
26
27class CourseNotFound(EnrollmentError):
28 pass
29
30
31class AlreadyEnrolled(EnrollmentError):
32 pass
33
34
35class CourseFull(EnrollmentError):
36 pass
37
38
39def check_enrollment(
40 student: Student | None,
41 course: Course | None,
42 already_enrolled: bool,
43 enrolled_count: int,
44) -> None:
45 if student is None:
46 raise StudentNotFound("学生不存在")
47 if course is None:
48 raise CourseNotFound("课程不存在")
49 if already_enrolled:
50 raise AlreadyEnrolled("该学生已报名这门课程")
51 if enrolled_count >= course.capacity:
52 raise CourseFull("课程已满")
规则失败时抛出明确的异常；“学生不存在”和“课程已满”不会再被笼统的 False 混在一起。domain 不导入 FastAPI 或 sqlite3，所以它是相对稳定的内层。
2. service：编排报名流程，声明存储需求
服务负责按顺序读学生、读课程、查重复与人数，然后调用领域规则，最后保存。EnrollmentStore 列出服务真正需要的五个操作。
1"""The enrollment use case and the storage interface it needs."""
2
3from typing import Protocol
4
5from .domain import Course, Student, check_enrollment
6
7
8class EnrollmentStore(Protocol):
9 def get_student(self, student_id: int) -> Student | None: ...
10
11 def get_course(self, course_id: int) -> Course | None: ...
12
13 def has_enrollment(self, student_id: int, course_id: int) -> bool: ...
14
15 def count_enrollments(self, course_id: int) -> int: ...
16
17 def add_enrollment(self, student_id: int, course_id: int) -> None: ...
18
19
20class EnrollmentService:
21 def __init__(self, store: EnrollmentStore) -> None:
22 self.store = store
23
24 def enroll(self, student_id: int, course_id: int) -> None:
25 student = self.store.get_student(student_id)
26 course = self.store.get_course(course_id)
27 already_enrolled = self.store.has_enrollment(student_id, course_id)
28 enrolled_count = self.store.count_enrollments(course_id)
29 check_enrollment(student, course, already_enrolled, enrolled_count)
30 self.store.add_enrollment(student_id, course_id)
测试这个服务时，可以给它传入一个内存实现；业务规则改变时，主要修改 domain；报名步骤改变时，主要修改 service。接口的意义不是把每个类都包装一次，而是让服务的存储需求具体、可替换。
3. db：实现这些存储操作
SQLite 版本负责建表、参数化查询与插入。示例启动时用 INSERT OR IGNORE 准备学生 1 和课程 1，课程容量为 2。
1"""SQLite implementation of the storage operations required by the service."""
2
3import sqlite3
4from contextlib import closing
5from pathlib import Path
6
7from .domain import AlreadyEnrolled, Course, Student
8
9
10class SQLiteEnrollmentStore:
11 def __init__(self, path: str | Path) -> None:
12 self.path = str(path)
13
14 def _connect(self) -> sqlite3.Connection:
15 conn = sqlite3.connect(self.path)
16 conn.execute("PRAGMA foreign_keys = ON")
17 return conn
18
19 def init_db(self) -> None:
20 with closing(self._connect()) as conn, conn:
21 conn.executescript("""
22 CREATE TABLE IF NOT EXISTS students (
23 id INTEGER PRIMARY KEY, name TEXT NOT NULL
24 );
25 CREATE TABLE IF NOT EXISTS courses (
26 id INTEGER PRIMARY KEY, title TEXT NOT NULL,
27 capacity INTEGER NOT NULL CHECK (capacity > 0)
28 );
29 CREATE TABLE IF NOT EXISTS enrollments (
30 student_id INTEGER NOT NULL REFERENCES students(id),
31 course_id INTEGER NOT NULL REFERENCES courses(id),
32 PRIMARY KEY (student_id, course_id)
33 );
34 """)
35 conn.execute(
36 "INSERT OR IGNORE INTO students (id, name) VALUES (?, ?)",
37 (1, "小明"),
38 )
39 conn.execute(
40 "INSERT OR IGNORE INTO courses (id, title, capacity) VALUES (?, ?, ?)",
41 (1, "Python 入门", 2),
42 )
43
44 def get_student(self, student_id: int) -> Student | None:
45 with closing(self._connect()) as conn:
46 row = conn.execute(
47 "SELECT id, name FROM students WHERE id = ?", (student_id,)
48 ).fetchone()
49 return Student(*row) if row else None
50
51 def get_course(self, course_id: int) -> Course | None:
52 with closing(self._connect()) as conn:
53 row = conn.execute(
54 "SELECT id, title, capacity FROM courses WHERE id = ?", (course_id,)
55 ).fetchone()
56 return Course(*row) if row else None
57
58 def has_enrollment(self, student_id: int, course_id: int) -> bool:
59 with closing(self._connect()) as conn:
60 row = conn.execute(
61 "SELECT 1 FROM enrollments WHERE student_id = ? AND course_id = ?",
62 (student_id, course_id),
63 ).fetchone()
64 return row is not None
65
66 def count_enrollments(self, course_id: int) -> int:
67 with closing(self._connect()) as conn:
68 row = conn.execute(
69 "SELECT COUNT(*) FROM enrollments WHERE course_id = ?", (course_id,)
70 ).fetchone()
71 return row[0]
72
73 def add_enrollment(self, student_id: int, course_id: int) -> None:
74 try:
75 with closing(self._connect()) as conn, conn:
76 conn.execute(
77 "INSERT INTO enrollments (student_id, course_id) VALUES (?, ?)",
78 (student_id, course_id),
79 )
80 except sqlite3.IntegrityError as exc:
81 # A duplicate may race with the earlier read. Other constraint
82 # failures should remain database errors instead of being mislabeled.
83 if self.has_enrollment(student_id, course_id):
84 raise AlreadyEnrolled("该学生已报名这门课程") from exc
85 raise
enrollments 表使用 (student_id, course_id) 作为主键。即使另一请求在“检查重复”和“插入”之间抢先写入，数据库仍会拒绝重复记录；适配器将这类冲突转换成领域异常。
这里还有一个刻意保留的边界：容量检查与插入没有放进同一事务。两个学生同时报名最后一个名额时，都可能看到“还剩一席”，最终超额。这个例子用于学习分层，不能直接当作高并发报名系统。实际系统应把人数检查和写入放进同一个受数据库锁或其他并发控制保护的事务，并对冲突做重试或明确返回。
4. api：映射 HTTP，并在外层装配
API 只把请求交给服务，再把可预期的失败映射到 404 或 409。创建数据库适配器、创建服务、把二者连接起来的几行，就是装配点。
1"""HTTP adapter and application assembly."""
2
3from fastapi import FastAPI, HTTPException
4from pydantic import BaseModel
5
6from .db import SQLiteEnrollmentStore
7from .domain import AlreadyEnrolled, CourseFull, CourseNotFound, StudentNotFound
8from .service import EnrollmentService
9
10
11class EnrollmentRequest(BaseModel):
12 student_id: int
13 course_id: int
14
15
16store = SQLiteEnrollmentStore("enrollment.db")
17store.init_db()
18service = EnrollmentService(store)
19app = FastAPI()
20
21
22@app.post("/enrollments", status_code=201)
23def enroll(request: EnrollmentRequest) -> dict[str, int]:
24 try:
25 service.enroll(request.student_id, request.course_id)
26 except (StudentNotFound, CourseNotFound) as exc:
27 raise HTTPException(status_code=404, detail=str(exc)) from exc
28 except (AlreadyEnrolled, CourseFull) as exc:
29 raise HTTPException(status_code=409, detail=str(exc)) from exc
30 return {"student_id": request.student_id, "course_id": request.course_id}
如果以后换成 PostgreSQL，主要工作是实现同一组存储操作，并在装配点把新实例传给 EnrollmentService。数据库迁移、事务语义和并发控制仍要认真处理；依赖注入只减少上层代码对某一种存储技术的绑定。
两个容易写错的细节
细节一：接口必须由「调用方」定义
回看代码：EnrollmentStore 这个接口，定义在 service.py 里，而不是 db.py 里。
1# service.py：谁需要，谁定义
2class EnrollmentStore(Protocol):
3 def get_student(self, student_id: int) -> Student | None: ...
4 def get_course(self, course_id: int) -> Course | None: ...
5 def has_enrollment(self, student_id: int, course_id: int) -> bool: ...
6 def count_enrollments(self, course_id: int) -> int: ...
7 def add_enrollment(self, student_id: int, course_id: int) -> None: ...
这一点是依赖倒置的核心，也最容易写错。假设反过来——接口定义在 db.py，service 去 from .db import EnrollmentStore——那么 service 依然依赖着 db 这个模块，只是从“依赖具体类”换成了“依赖某个模块里的接口”而已。外层依旧牵着内层走，改动 db 仍会波及 service。
所以有一条判据：谁来定义接口，谁就掌握主动权。接口应该由需要它的内层（调用方）定义，由外层（实现方）去满足。
用餐厅类比：是“大堂经理”写下岗位说明书——我需要一个能查数据、能写数据的仓管员——然后“仓库管理员”照着这份说明书来应聘。而不是仓管员自己写一份说明书塞给经理。
细节二：抽象在 Python 里有三种写法
“依赖抽象”要真正落地，Python 给了三档选择：
| 方式 | 写法 | 是否需要继承 | 运行时检查 |
|---|---|---|---|
| 鸭子类型 | 什么都不写，直接传对象 | 不需要 | 无（调用到时才报错） |
Protocol | class X(Protocol) | 不需要，方法签名匹配即可 | 加 @runtime_checkable 才有 |
ABC + @abstractmethod | class X(ABC) | 必须显式继承 | 有（抽象类不能实例化） |
三档都能切断对具体实现的依赖，区别只在“约束有多强”。
ABC（Abstract Base Class，抽象基类）是最传统的方式，用 @abstractmethod 声明子类必须实现的方法。它属于名义类型（nominal typing）：子类必须显式写 class SQLiteEnrollmentStore(EnrollmentStore)，否则不算数。
Protocol 是 Python 3.8+ 引入的结构化类型（structural typing）。注意看 db.py，它根本没有继承 EnrollmentStore：
1# db.py：一个有五个方法的普通类，不需要继承任何接口
2class SQLiteEnrollmentStore:
3 def get_student(self, student_id: int) -> Student | None: ...
4 # ...另外四个方法
它靠“长得像不像”来判断，而不是“是不是亲生的”。本文选 Protocol 正是因为这一点：db 完全不需要知道 service 定义过这样一个接口，只要五个方法齐备，它就是 EnrollmentStore。
“虚函数（virtual function）”是 C++ 里实现运行时多态的机制（虚函数表）；Python 没有 virtual 关键字，所有方法天然就是虚函数。所以“用虚函数解决依赖”并不准确——虚函数是底层语言机制，依赖倒置是架构原则，两者不在一个层次。换到 Java 叫 interface，换到 Go 也叫 interface，但“接口归调用方所有”这条规则，在哪门语言里都一样。
本地运行与验证
在 registration_demo 目录执行：
1python3 -m venv .venv
2source .venv/bin/activate
3python -m pip install "fastapi[standard]"
4python -m uvicorn registration_layers.api:app --reload
另开终端发送第一次报名请求：
1curl -i -X POST http://127.0.0.1:8000/enrollments \
2 -H 'Content-Type: application/json' \
3 -d '{"student_id":1,"course_id":1}'
预期得到 201 Created 和 {"student_id":1,"course_id":1}。原样再请求一次，预期得到 409 Conflict 与“已报名”；把 student_id 改成 999，预期得到 404 Not Found。删除当前目录下的 enrollment.db 可以重置演示数据。
不启动 Web 服务也能直接检查核心规则：
1python - <<'PY'
2from registration_layers.domain import (
3 AlreadyEnrolled, Course, CourseFull, Student, check_enrollment,
4)
5
6student = Student(1, "小明")
7course = Course(1, "Python 入门", 2)
8check_enrollment(student, course, False, 1) # 允许报名
9
10for error, enrolled, count in [
11 (AlreadyEnrolled, True, 1),
12 (CourseFull, False, 2),
13]:
14 try:
15 check_enrollment(student, course, enrolled, count)
16 except error:
17 print(error.__name__, "验证通过")
18PY
这正是分层的直接收益：规则测试不需要 HTTP 和数据库；要测试流程，可以传入一个实现了 EnrollmentStore 五个方法的内存对象；要验证数据库，再单独连接 SQLite。
什么时候值得这样拆
如果只有一个很短、不会变化的脚本，分四层会增加文件和跳转成本。报名流程有多条规则、可能换 Web 框架或数据库、需要独立测试时，这种拆分才开始回本。判断标准不是“文件是否足够多”，而是每次需求变化能否找到主要修改位置，并且内层规则能否脱离外部工具运行。