Python 基础体系 · 第 75/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。
Python 应用架构：模块边界、依赖方向、领域层和可替换适配器
一个 Python 应用开始变复杂，通常不是因为某个函数太长，而是因为变化被放在了错误的边界上：
- HTTP 参数校验、数据库查询和业务规则写在同一个函数里；
- 领域对象直接调用 ORM、消息队列或第三方 SDK；
- 测试必须启动数据库才能验证一个金额计算；
- 更换数据库、支付供应商或 Web 框架时，需要修改大量业务代码；
- 包之间相互导入，最终出现循环导入、隐式初始化和难以定位的启动错误。
应用架构要解决的不是“目录如何排列”，而是三个更基础的问题：
- 哪个模块负责哪个决策？
- 哪些模块允许依赖哪些模块？
- 外部系统的变化如何被限制在适配器内部？
本文以一个“创建订单”的应用为例，逐步建立模块边界、依赖方向、领域层和可替换适配器，并说明这些设计如何与 Web API 工程中的契约、错误、幂等、限流和版本，以及生产交付中的配置、迁移、灰度和回滚衔接。
一、先区分模块、包、层和边界
1. 模块是什么
在 Python 中，模块通常指一个可导入的 Python 文件；包则是用于组织模块层次的模块结构。导入一个模块时，Python 会先搜索模块，再把搜索结果绑定到当前作用域；模块首次导入时会创建并初始化模块对象，找不到目标时抛出 ModuleNotFoundError。(docs.python.org)
例如：
shop/
├── orders/
│ ├── domain.py
│ ├── application.py
│ └── repository.py
└── adapters/
└── sqlite_orders.py
这里：
orders.domain是一个模块；orders是一个包；orders这一组模块可以被视为一个功能模块；orders.domain、orders.application等文件之间形成了代码依赖；orders与adapters之间则应形成架构依赖关系。
因此，“模块边界”不只是文件边界。
2. 模块边界是什么
模块边界是一个模块对外暴露的稳定能力，以及它刻意隐藏的实现细节。
一个边界至少包含四部分：
调用者 ──输入──> 模块公开接口
调用者 <─输出── 模块公开接口
调用者 <─错误── 模块公开错误语义
模块内部 ────── 隐藏实现
如果调用方必须知道：
- 数据库表名；
- ORM 的查询对象；
- 第三方支付 SDK 的异常类型；
- HTTP 框架的请求对象；
- 配置文件的具体结构；
才能调用一个业务能力，那么边界通常已经泄漏。
边界的判断标准不是“这个类是否写在另一个文件”，而是：
调用方是否可以在不了解内部实现的情况下使用该能力？
例如，下面两个接口的边界质量不同：
# 泄漏数据库实现
def create_order(session, user_id: int, rows: list[dict]) -> dict:
...
# 隐藏数据库实现
def create_order(command: CreateOrder) -> Order:
...
第一个接口把数据库会话暴露给调用方，导致业务服务依赖 ORM 生命周期。第二个接口只表达业务输入和业务输出，具体数据如何保存由其他组件决定。
3. 层不是目录，而是允许的依赖集合
常见的分层可以表示为：
接口层 HTTP、CLI、消息消费者
应用层 用例编排、事务边界、权限和幂等协作
领域层 业务规则、实体、值对象、领域错误
适配器层 数据库、HTTP 客户端、消息队列、文件系统
基础设施层 连接池、配置、日志、进程启动
“层”真正约束的是谁可以调用谁，而不是目录名称。
例如：
错误方向：
domain ──> sqlalchemy
domain ──> fastapi
domain ──> stripe_sdk
更稳定的方向：
interface ──> application ──> domain
↑
adapter ── implements ────────┘
领域层知道“需要一个订单仓储”，但不应该知道仓储是 SQLite、PostgreSQL 还是内存字典。
二、依赖方向：把业务决策放在内侧
1. 什么是依赖
在架构讨论中，“依赖”不只表示 import。
如果模块 A 的正确运行必须知道模块 B 的类型、生命周期、异常或实现细节，那么 A 依赖 B。
下面三种依赖都是真实依赖：
from sqlalchemy.orm import Session # 类型依赖
raise requests.Timeout(...) # 异常依赖
client = StripeClient(api_key) # 生命周期和实现依赖
即使通过参数传入对象，依赖也没有消失：
def pay(session: Session, client: StripeClient) -> None:
...
它只是从全局依赖变成了显式依赖。
2. 依赖方向的形式化条件
设模块集合为：
如果模块 需要模块 的实现、类型或错误语义，则记为：
好的应用架构通常要求：
而不是：
以订单创建为例：
- 业务规则：订单至少包含一行商品，商品数量必须为正；
- 外部细节：订单最终写入 SQLite；
- 业务规则变化频率通常与数据库驱动变化无关。
如果领域层直接依赖 SQLite，那么数据库细节变化会沿依赖边传播到业务规则中：
domain -> sqlite -> database
更合理的关系是：
domain 定义 OrderRepository 接口
sqlite 依赖并实现 OrderRepository
application 使用 OrderRepository
依赖图变成：
application ──> domain
adapter ──────> domain
interface ────> application
bootstrap ────> interface + adapter
注意：这并不意味着所有箭头都“向上”。适配器依赖领域接口，是因为它需要满足领域定义的协议。
3. 为什么依赖倒置不是“所有东西都抽象化”
依赖倒置的核心不是增加接口数量，而是把稳定的业务约束放在更内层，把容易变化的技术实现放在外层。
如果某个实现只有一种、变化频率低、也不需要隔离测试，就没有必要为了形式上的“可替换”创建抽象。
例如：
def normalize_email(value: str) -> str:
return value.strip().casefold()
这个函数没有数据库、网络或进程边界，直接使用即可。
相反，下面这些依赖通常值得隔离：
- 数据库；
- 外部 HTTP 服务；
- 当前时间；
- 随机数；
- 文件系统；
- 消息代理；
- 进程环境变量；
- 第三方支付或搜索 SDK。
原因不是它们“不优雅”，而是它们有独立的失败模式、生命周期和替换成本。
三、领域层：只保存业务不变量
1. 领域层的定义
领域层是表达业务概念和业务规则的代码集合。它关注的是：
- 什么是合法的订单；
- 什么状态可以转换到什么状态；
- 金额如何计算；
- 哪些操作违反业务规则；
- 哪些事实需要被记录。
领域层不负责：
- 解析 HTTP JSON；
- 查询数据库；
- 读取环境变量；
- 重试网络请求；
- 选择线程池或进程模型；
- 把异常转换成 HTTP 响应。
领域层的核心是不变量。
不变量是指：在对象对外可见的整个生命周期中，必须始终成立的条件。
例如，订单可以定义为：
其中：
- 是订单行数；
unit_price是单价；quantity是购买数量；total是订单总额。
如果订单进入 paid 状态，则不能再次修改商品行。于是还存在状态不变量：
2. 值对象：让非法值难以进入系统
值对象是由值定义身份、通常没有独立生命周期的对象，例如金额、邮箱、订单号。
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
class DomainError(Exception):
"""所有可预期的领域规则错误。"""
@dataclass(frozen=True, slots=True)
class Money:
amount: Decimal
currency: str = "CNY"
def __post_init__(self) -> None:
if self.amount < 0:
raise DomainError("金额不能为负数")
if len(self.currency) != 3:
raise DomainError("货币代码必须是三个字符")
def add(self, other: Money) -> Money:
if self.currency != other.currency:
raise DomainError("不能相加不同货币")
return Money(self.amount + other.amount, self.currency)
def multiply(self, quantity: int) -> Money:
if quantity <= 0:
raise DomainError("数量必须为正数")
return Money(self.amount * quantity, self.currency)
@dataclass 可以根据类型注解自动生成诸如 __init__() 和 __repr__() 等特殊方法；这里使用 frozen=True 和 slots=True 是设计选择，不代表数据类天然不可变或天然适合领域建模。(docs.python.org)
该类的关键不在装饰器，而在不变量：
- 构造
Money时禁止负数； - 相加前检查货币一致；
- 乘法前检查数量为正；
- 通过返回新对象，避免调用者绕过规则修改内部状态。
反例是直接使用裸 Decimal：
price = Decimal("10.00")
quantity = -3
total = price * quantity
这段代码在 Python 层面可以运行，但它让“负数量是否允许”变成了每个调用点都必须记住的隐含规则。调用点越多，规则越容易不一致。
3. 实体：身份和状态变化
订单是实体，因为两个订单即使内容完全相同，也可能由不同的订单 ID 区分。
from dataclasses import dataclass, field
from enum import StrEnum
class OrderStatus(StrEnum):
DRAFT = "draft"
PAID = "paid"
CANCELLED = "cancelled"
@dataclass(slots=True)
class OrderLine:
sku: str
unit_price: Money
quantity: int
def __post_init__(self) -> None:
if not self.sku:
raise DomainError("商品 SKU 不能为空")
if self.quantity <= 0:
raise DomainError("商品数量必须为正数")
@property
def subtotal(self) -> Money:
return self.unit_price.multiply(self.quantity)
@dataclass(slots=True)
class Order:
order_id: str
user_id: int
lines: list[OrderLine] = field(default_factory=list)
status: OrderStatus = OrderStatus.DRAFT
def __post_init__(self) -> None:
if not self.order_id:
raise DomainError("订单 ID 不能为空")
if self.user_id <= 0:
raise DomainError("用户 ID 必须为正数")
if not self.lines:
raise DomainError("订单至少包含一行商品")
@property
def total(self) -> Money:
result = Money(Decimal("0.00"))
for line in self.lines:
result = result.add(line.subtotal)
return result
def mark_paid(self) -> None:
if self.status is not OrderStatus.DRAFT:
raise DomainError(f"{self.status} 订单不能标记为已支付")
self.status = OrderStatus.PAID
def cancel(self) -> None:
if self.status is OrderStatus.PAID:
raise DomainError("已支付订单不能直接取消")
if self.status is OrderStatus.CANCELLED:
raise DomainError("订单已经取消")
self.status = OrderStatus.CANCELLED
状态变化必须经过实体方法，而不是由外部代码随意赋值：
# 不推荐：外部代码绕过状态规则
order.status = OrderStatus.PAID
# 推荐：由实体验证状态转换
order.mark_paid()
如果“支付后不能取消”只写在 API 处理函数中，那么订单还可能从定时任务、后台脚本或消息消费者中被非法修改。规则放在实体中，所有入口共享同一约束。
4. 领域层不等于“所有业务代码都写在实体里”
复杂系统中，规则可能跨越多个实体。例如：
- 检查用户信用额度；
- 比较库存和订单数量；
- 计算多个订单的折扣；
- 根据时间段选择价格策略。
这类规则可以放在领域服务中：
class CreditLimitExceeded(DomainError):
pass
def ensure_credit_available(
current_debt: Money,
new_order_total: Money,
limit: Money,
) -> None:
if current_debt.currency != limit.currency:
raise DomainError("信用额度货币不一致")
projected = current_debt.add(new_order_total)
if projected.amount > limit.amount:
raise CreditLimitExceeded("超过信用额度")
领域服务仍然只处理业务概念。它不应该接收 Session、Request 或 SDK 客户端。
四、端口和适配器：把外部世界隔离出去
1. 端口是什么
端口是内层对外部能力提出的抽象要求。
例如，应用层需要“保存订单”，但并不需要知道保存动作由什么数据库完成：
from typing import Protocol
class OrderRepository(Protocol):
def get(self, order_id: str) -> Order | None:
...
def save(self, order: Order) -> None:
...
Protocol 用于描述结构化接口：实现类不一定必须显式继承该协议，只要提供兼容的方法和属性，就可以被静态类型检查器视为满足该接口。Python 的 typing 文档将 Protocol 作为类型提示机制的一部分；若使用 runtime_checkable()，运行时检查主要关注属性是否存在，并不会验证完整的类型签名。(docs.python.org)
这里使用 Protocol 有两个重要效果：
- 领域和应用层只依赖
OrderRepository的行为； - 测试可以传入一个简单的内存实现，而不需要构造数据库对象。
也可以使用 abc.ABC 和 @abstractmethod。抽象基类会限制未实现抽象方法的类被实例化，并支持显式继承关系；这与 Protocol 的结构化类型检查不同。(docs.python.org)
选择原则可以这样理解：
只需要约定行为、希望测试替换简单：Protocol
需要运行时抽象基类、共享实现或显式继承体系：ABC
不要为了“面向对象”同时定义 Protocol、ABC、第三方接口和一层包装类。抽象的数量应该由变化边界决定。
2. 适配器是什么
适配器是把外部系统的接口转换为端口接口的实现。
class InMemoryOrderRepository:
def __init__(self) -> None:
self._orders: dict[str, Order] = {}
def get(self, order_id: str) -> Order | None:
return self._orders.get(order_id)
def save(self, order: Order) -> None:
self._orders[order.order_id] = order
它满足 OrderRepository 的方法形状，因此应用层可以使用它。
数据库适配器可能是：
class SqliteOrderRepository:
def __init__(self, connection) -> None:
self._connection = connection
def get(self, order_id: str) -> Order | None:
row = self._connection.execute(
"""
SELECT order_id, user_id, status
FROM orders
WHERE order_id = ?
""",
(order_id,),
).fetchone()
if row is None:
return None
lines = [
OrderLine(
sku=item_row[0],
unit_price=Money(Decimal(item_row[1])),
quantity=item_row[2],
)
for item_row in self._connection.execute(
"""
SELECT sku, unit_price, quantity
FROM order_lines
WHERE order_id = ?
ORDER BY id
""",
(order_id,),
)
]
return Order(
order_id=row[0],
user_id=row[1],
lines=lines,
status=OrderStatus(row[2]),
)
def save(self, order: Order) -> None:
self._connection.execute(
"""
INSERT INTO orders(order_id, user_id, status)
VALUES (?, ?, ?)
ON CONFLICT(order_id) DO UPDATE SET status = excluded.status
""",
(order.order_id, order.user_id, order.status.value),
)
self._connection.execute(
"DELETE FROM order_lines WHERE order_id = ?",
(order.order_id,),
)
self._connection.executemany(
"""
INSERT INTO order_lines(order_id, sku, unit_price, quantity)
VALUES (?, ?, ?, ?)
""",
[
(
order.order_id,
line.sku,
str(line.unit_price.amount),
line.quantity,
)
for line in order.lines
],
)
这里的 SQLite SQL 是适配器内部细节。应用层只看到：
repo.get(order_id)
repo.save(order)
如果未来从 SQLite 切换到 PostgreSQL，主要变化应集中在适配器和启动组装代码，而不是 Order.mark_paid() 或订单总额计算。
3. 端口不是数据库 CRUD 的机械翻译
下面这种接口虽然也叫端口，但抽象层次偏低：
class OrderRepository(Protocol):
def insert_row(self, table: str, values: dict) -> None:
...
它把数据库概念泄漏给内层。领域层真正需要的是业务能力：
class OrderRepository(Protocol):
def get(self, order_id: str) -> Order | None:
...
def save(self, order: Order) -> None:
...
端口应由调用者需要什么决定，而不是由适配器提供什么决定。
五、应用层：编排用例，而不是承载所有规则
领域层负责局部业务不变量，应用层负责一次用例中的协作过程。
例如“创建订单”需要：
- 接收已规范化的命令；
- 生成订单 ID；
- 构造领域对象；
- 保存订单；
- 返回领域结果。
from dataclasses import dataclass
from typing import Callable
@dataclass(frozen=True, slots=True)
class CreateOrderLine:
sku: str
unit_price: Decimal
quantity: int
@dataclass(frozen=True, slots=True)
class CreateOrder:
user_id: int
lines: tuple[CreateOrderLine, ...]
class OrderIdGenerator(Protocol):
def __call__(self) -> str:
...
class CreateOrderService:
def __init__(
self,
repository: OrderRepository,
generate_order_id: OrderIdGenerator,
) -> None:
self._repository = repository
self._generate_order_id = generate_order_id
def execute(self, command: CreateOrder) -> Order:
order = Order(
order_id=self._generate_order_id(),
user_id=command.user_id,
lines=[
OrderLine(
sku=line.sku,
unit_price=Money(line.unit_price),
quantity=line.quantity,
)
for line in command.lines
],
)
self._repository.save(order)
return order
应用层没有直接调用 uuid.uuid4()，而是接收一个 ID 生成器。这样做不是为了把每个函数都变成高阶函数，而是因为 ID 生成会影响：
- 幂等键设计；
- 测试稳定性；
- 数据迁移；
- 分布式追踪；
- 失败重试后的行为。
一个固定生成器即可测试：
repo = InMemoryOrderRepository()
service = CreateOrderService(repo, lambda: "order-001")
order = service.execute(
CreateOrder(
user_id=42,
lines=(
CreateOrderLine("book", Decimal("39.90"), 2),
),
)
)
assert order.total == Money(Decimal("79.80"))
assert repo.get("order-001") is order
执行过程中的中间结果是：
输入：
user_id = 42
sku = book
unit_price = 39.90
quantity = 2
构造 OrderLine：
subtotal = 39.90 × 2 = 79.80
构造 Order：
status = draft
total = 79.80
保存：
repo["order-001"] = Order(...)
如果数量为 0，错误发生在 OrderLine 构造阶段；如果订单没有商品行，错误发生在 Order 构造阶段。错误越接近非法状态产生的位置，诊断越直接。
六、事务边界、幂等和错误翻译
1. 领域层不能负责数据库事务
Order 可以判断“已支付订单不能修改”，但它不能提交数据库事务。事务属于外部资源的生命周期，应由应用层或基础设施层控制。
一个更完整的端口可能包含工作单元：
class UnitOfWork(Protocol):
orders: OrderRepository
def __enter__(self) -> UnitOfWork:
...
def __exit__(self, exc_type, exc_value, traceback) -> None:
...
def commit(self) -> None:
...
def rollback(self) -> None:
...
应用服务：
class PayOrderService:
def __init__(self, uow: UnitOfWork) -> None:
self._uow = uow
def execute(self, order_id: str) -> Order:
with self._uow:
order = self._uow.orders.get(order_id)
if order is None:
raise DomainError("订单不存在")
order.mark_paid()
self._uow.orders.save(order)
self._uow.commit()
return order
这里有一个关键故障路径：
读取订单
│
├─ 订单不存在 ──> 返回业务错误
│
├─ 状态不允许 ──> 领域错误，不提交
│
├─ 保存失败 ────> 基础设施错误，回滚
│
└─ 提交成功 ────> 返回已支付订单
如果 save() 已执行但 commit() 失败，应用层不能假设数据已经持久化。接口层也不能把所有异常都转换成 400 Bad Request：
DomainError -> 400 或 409
ValidationError -> 400
RepositoryUnavailable -> 503
UnexpectedError -> 500
具体 HTTP 状态码属于接口层契约，不应反向污染领域层。
2. 幂等必须跨越应用层和适配器层
假设客户端因超时重试创建订单：
请求 1：创建订单 order-001
服务端：已保存，但响应丢失
请求 2：再次创建订单 order-001
如果 ID 由服务端每次随机生成，两次请求可能创建两个订单。应用层需要接收幂等键：
@dataclass(frozen=True, slots=True)
class CreateOrder:
user_id: int
idempotency_key: str
lines: tuple[CreateOrderLine, ...]
然后端口提供幂等记录：
class IdempotencyStore(Protocol):
def get(self, key: str) -> Order | None:
...
def save(self, key: str, order: Order) -> None:
...
流程为：
读取幂等键
│
├─ 已存在 ──> 返回原订单
│
└─ 不存在
│
├─ 构造订单
├─ 保存订单
└─ 保存幂等键
但这仍然存在并发竞争：
请求 A：检查 key，不存在
请求 B：检查 key，不存在
请求 A：创建订单
请求 B：创建订单
因此，真正的幂等保证通常需要数据库唯一约束、原子插入或分布式锁。仅在 Python 字典中先查后写，只能在单线程、单进程且没有并发的演示场景中成立。
这说明了模块边界的限制：
架构可以隔离幂等策略，但不能用接口抽象代替底层并发原子性。
3. 限流也不应写进领域实体
限流是接口或基础设施关注点：
- 按用户、IP、令牌或租户计数；
- 维护时间窗口；
- 在多个进程之间共享状态；
- 产生
429响应和Retry-After。
订单实体不应该出现：
order.check_ip_rate_limit(...)
因为限流不是订单不变量。把它放进领域层会让每个领域对象都依赖请求上下文和外部计数器。
七、接口层与领域层之间要有翻译
接口层负责把外部格式翻译成应用命令，再把应用结果翻译成外部响应。
def handle_create_order(payload: dict, service: CreateOrderService) -> dict:
try:
command = CreateOrder(
user_id=int(payload["user_id"]),
lines=tuple(
CreateOrderLine(
sku=str(item["sku"]),
unit_price=Decimal(str(item["unit_price"])),
quantity=int(item["quantity"]),
)
for item in payload["lines"]
),
)
order = service.execute(command)
except KeyError as exc:
return {"status": 400, "body": {"error": f"缺少字段: {exc.args[0]}"}}
except DomainError as exc:
return {"status": 409, "body": {"error": str(exc)}}
return {
"status": 201,
"body": {
"order_id": order.order_id,
"user_id": order.user_id,
"status": order.status.value,
"total": str(order.total.amount),
"currency": order.total.currency,
},
}
该函数不是 Web 框架专属代码，因此可以直接运行：
repo = InMemoryOrderRepository()
service = CreateOrderService(repo, lambda: "order-001")
response = handle_create_order(
{
"user_id": 42,
"lines": [
{"sku": "book", "unit_price": "39.90", "quantity": 2},
],
},
service,
)
print(response)
预期结果：
{
"status": 201,
"body": {
"order_id": "order-001",
"user_id": 42,
"status": "draft",
"total": "79.80",
"currency": "CNY",
},
}
接口层完成了三种转换：
JSON 字符串
-> Decimal / int
-> CreateOrder
-> Order
-> JSON 可表达的 dict
如果把 JSON 字段名、HTTP 状态码直接塞进领域对象，领域层就会被一种接口协议绑定。例如：
class Order:
def to_http_response(self) -> dict:
...
这会让 CLI、消息消费者和后台任务也被迫使用 HTTP 语义。
八、包结构：让依赖关系可见
一个小型项目可以使用如下结构：
shop/
├── pyproject.toml
└── src/
└── shop/
├── __init__.py
├── domain/
│ ├── __init__.py
│ └── orders.py
├── application/
│ ├── __init__.py
│ └── orders.py
├── adapters/
│ ├── __init__.py
│ └── memory_orders.py
└── interfaces/
├── __init__.py
└── http_orders.py
建议的导入方向：
shop.interfaces.http_orders
│
▼
shop.application.orders
│
▼
shop.domain.orders
shop.adapters.memory_orders
│
▼
shop.domain.orders
领域包不应导入：
from shop.adapters.memory_orders import InMemoryOrderRepository
from fastapi import Request
from sqlalchemy import select
启动代码则负责组装具体实现：
from shop.adapters.memory_orders import InMemoryOrderRepository
from shop.application.orders import CreateOrderService
from shop.interfaces.http_orders import handle_create_order
repository = InMemoryOrderRepository()
service = CreateOrderService(repository, lambda: "order-001")
这一步称为组合根：应用在一个明确位置决定使用哪些实现。
不要让业务模块自行决定适配器：
# 不推荐
class CreateOrderService:
def __init__(self) -> None:
self.repository = SqliteOrderRepository(open_connection())
这样会造成：
- 测试无法替换仓储；
- 连接创建时机不可控；
- 配置读取散落在业务代码；
- 每个服务都重复构造基础设施；
- 关闭连接的责任不清楚。
Python 导入系统会缓存已导入模块；sys.modules 保存已经导入的模块对象，重复导入通常复用缓存，但导入时的模块初始化和顶层副作用仍然会影响启动行为。(docs.python.org)
因此，__init__.py 和模块顶层代码应尽量保持轻量：
# 风险较高：导入时连接数据库
connection = create_database_connection()
# 风险较高：导入时读取并校验生产密钥
settings = load_production_settings()
更可控的方式是显式启动：
def build_application() -> CreateOrderService:
settings = load_settings()
connection = create_connection(settings.database_url)
repository = SqliteOrderRepository(connection)
return CreateOrderService(repository, generate_order_id)
导入模块只定义类和函数，资源在启动生命周期中创建，在关闭生命周期中释放。
九、循环依赖为什么是边界错误
考虑下面的结构：
domain/orders.py
imports application/orders.py
application/orders.py
imports domain/orders.py
当 Python 加载 domain.orders 时，它还没有执行完，就进入 application.orders；后者又尝试从尚未初始化完成的 domain.orders 导入名称，于是可能出现：
ImportError: cannot import name 'Order' from partially initialized module ...
循环导入不是单纯的“import 写法问题”，而是两个模块互相需要对方的实现，说明边界没有确定。
常见修复步骤：
- 找出循环边；
- 判断两个模块中哪个概念更稳定；
- 把共享的纯领域概念移到领域模块；
- 把编排逻辑留在应用模块；
- 让接口层只依赖应用层；
- 不用“延迟导入”掩盖真实的双向依赖。
延迟导入有时能解决启动时序问题：
def handler():
from shop.application.orders import CreateOrderService
...
但它不改变架构依赖图。若两个模块仍然互相调用，问题只会从启动时暴露推迟到运行时暴露。
十、可替换适配器的完整端到端示例
下面用同一套应用服务分别连接内存适配器和 SQLite 适配器。
1. 内存适配器
class InMemoryOrderRepository:
def __init__(self) -> None:
self._orders: dict[str, Order] = {}
def get(self, order_id: str) -> Order | None:
return self._orders.get(order_id)
def save(self, order: Order) -> None:
self._orders[order.order_id] = order
适合：
- 领域测试；
- 应用服务单元测试；
- 本地演示；
- 不需要持久化的短生命周期任务。
不适合直接宣称支持生产级并发、崩溃恢复或跨进程一致性。
2. SQLite 适配器的初始化
import sqlite3
def initialize_database(connection: sqlite3.Connection) -> None:
connection.executescript(
"""
PRAGMA foreign_keys = ON;
CREATE TABLE IF NOT EXISTS orders (
order_id TEXT PRIMARY KEY,
user_id INTEGER NOT NULL,
status TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS order_lines (
id INTEGER PRIMARY KEY AUTOINCREMENT,
order_id TEXT NOT NULL,
sku TEXT NOT NULL,
unit_price TEXT NOT NULL,
quantity INTEGER NOT NULL,
FOREIGN KEY (order_id) REFERENCES orders(order_id)
);
"""
)
connection.commit()
初始化数据库不是领域层职责。它属于迁移或基础设施启动流程。
生产环境不应把结构迁移随意放进每次应用启动中，原因包括：
- 多个进程同时启动；
- 迁移耗时导致健康检查失败；
- 回滚需要明确的兼容策略；
- 数据库权限可能不允许应用进程修改结构。
更安全的流程是：
构建迁移
-> 在发布阶段执行迁移
-> 验证迁移成功
-> 启动新版本
-> 灰度观察
-> 失败则回滚应用版本
但数据库回滚不总是等价于应用回滚。删除列、改变枚举语义等破坏性迁移，可能使旧应用无法读取新结构。因此应优先使用“向前兼容”的两阶段迁移：
阶段 1：增加新字段，旧字段仍保留
阶段 2：应用同时写旧字段和新字段
阶段 3：切换读取逻辑
阶段 4：确认无旧版本后删除旧字段
3. 替换适配器
def build_memory_service() -> CreateOrderService:
repository = InMemoryOrderRepository()
return CreateOrderService(repository, lambda: "memory-001")
def build_sqlite_service() -> CreateOrderService:
connection = sqlite3.connect(":memory:")
initialize_database(connection)
repository = SqliteOrderRepository(connection)
return CreateOrderService(repository, lambda: "sqlite-001")
两个服务都使用同一个应用层：
command = CreateOrder(
user_id=42,
lines=(
CreateOrderLine("book", Decimal("39.90"), 2),
),
)
memory_order = build_memory_service().execute(command)
sqlite_order = build_sqlite_service().execute(command)
assert memory_order.total == sqlite_order.total
assert memory_order.status == sqlite_order.status
这并不证明两个适配器完全等价。还需要验证：
- 重复保存是否幂等；
- 事务失败是否回滚；
- 并发写入是否安全；
- 读取后领域对象是否保留完整状态；
- 数据库精度和时区是否正确；
- 适配器异常是否被转换为应用可理解的错误。
可替换性是一个行为契约，而不是“方法名相同”。
十一、适配器契约测试比单纯模拟更可靠
如果只有单元测试：
service = CreateOrderService(fake_repository, fake_id_generator)
测试可能只证明服务会调用假对象，却没有证明真实数据库适配器能正确保存和恢复订单。
可以定义适配器共享测试：
def repository_contract(make_repository) -> None:
repository = make_repository()
order = Order(
order_id="order-001",
user_id=42,
lines=[
OrderLine(
sku="book",
unit_price=Money(Decimal("39.90")),
quantity=2,
)
],
)
assert repository.get("order-001") is None
repository.save(order)
loaded = repository.get("order-001")
assert loaded is not None
assert loaded.order_id == order.order_id
assert loaded.user_id == order.user_id
assert loaded.total == order.total
assert loaded.status == OrderStatus.DRAFT
然后分别运行：
repository_contract(lambda: InMemoryOrderRepository())
repository_contract(make_sqlite_repository)
契约测试验证的是端口的行为，而不是某个实现的内部代码。
但契约也要有边界。InMemoryOrderRepository 如果不模拟事务和并发，它不能替代集成测试。合理的测试分层是：
领域单元测试 验证不变量和状态转换
应用单元测试 验证用例编排和错误传播
适配器契约测试 验证端口行为
集成测试 验证真实数据库、网络和事务
接口测试 验证 HTTP 契约、状态码和错误格式
十二、异步边界：不要因为使用 async 就改变领域模型
Web API 可能使用异步框架，数据库驱动也可能提供异步接口，但“业务规则是否允许支付后取消”并不会因为调用方式变成异步而改变。
可以让端口表达异步能力：
from typing import Protocol
class AsyncOrderRepository(Protocol):
async def get(self, order_id: str) -> Order | None:
...
async def save(self, order: Order) -> None:
...
应用服务相应变为：
class AsyncPayOrderService:
def __init__(self, repository: AsyncOrderRepository) -> None:
self._repository = repository
async def execute(self, order_id: str) -> Order:
order = await self._repository.get(order_id)
if order is None:
raise DomainError("订单不存在")
order.mark_paid()
await self._repository.save(order)
return order
这里需要区分两件事：
异步调用协议：如何等待外部 I/O
领域规则：哪些状态和数值是合法的
不要把 async 扩散到没有 I/O 的领域函数：
# 没有外部等待，不需要异步
def mark_paid(self) -> None:
...
如果领域层为了“和 Web 层保持一致”全部使用异步，代码会增加调度复杂度，却没有获得并发收益。
十三、配置、日志和时间依赖的边界
1. 配置应在组合根读取
不推荐：
# domain/orders.py
import os
MAX_ORDER_LINES = int(os.environ["MAX_ORDER_LINES"])
这会使领域规则绑定到进程环境，并且在导入时就可能因配置缺失而失败。
更清晰的方式是把配置转换为领域所需的值：
@dataclass(frozen=True, slots=True)
class OrderPolicy:
max_lines: int
def __post_init__(self) -> None:
if self.max_lines <= 0:
raise ValueError("max_lines 必须为正数")
然后由应用服务或领域服务接收：
class CreateOrderService:
def __init__(
self,
repository: OrderRepository,
generate_order_id: OrderIdGenerator,
policy: OrderPolicy,
) -> None:
self._repository = repository
self._generate_order_id = generate_order_id
self._policy = policy
配置文件格式、环境变量名称和密钥加载属于基础设施；OrderPolicy(max_lines=50) 才是领域可理解的配置。
2. 时间必须显式注入
直接调用系统时间会让测试依赖真实时钟：
from datetime import datetime, timezone
created_at = datetime.now(timezone.utc)
如果订单过期规则需要可重复测试，可以定义时钟端口：
from datetime import datetime
from typing import Protocol
class Clock(Protocol):
def now(self) -> datetime:
...
class FixedClock:
def __init__(self, value: datetime) -> None:
self._value = value
def now(self) -> datetime:
return self._value
这不是为了抽象所有标准库调用，而是为了隔离一个会影响业务结果的外部输入。
3. 日志不是领域事件
领域层可以产生领域事件：
@dataclass(frozen=True, slots=True)
class OrderPaid:
order_id: str
但不应直接：
logger.info("order paid", extra={"order_id": order.order_id})
原因是日志格式、采样、输出位置和关联请求 ID 属于运行环境。应用层可以在用例完成后记录日志，或由事件发布适配器处理。
十四、常见失败结构及诊断方法
1. “万能 service.py”
service.py
├── 解析 HTTP
├── 查询数据库
├── 校验业务规则
├── 调用支付 SDK
├── 发送邮件
└── 组装 JSON
失败表现：
- 一个函数有多个变化原因；
- 单元测试需要大量 mock；
- 异常类型混杂；
- 数据库事务和网络重试互相影响；
- 修改响应字段可能破坏业务逻辑。
诊断方法是沿着代码逐句标注责任：
读取 request.json -> 接口层
校验数量 > 0 -> 领域层
查询订单 -> 仓储适配器
决定是否允许支付 -> 领域层
调用支付供应商 -> 外部服务适配器
返回 HTTP 201 -> 接口层
如果一个函数同时拥有五种颜色的责任，它通常需要拆分。
2. “仓储接口”泄漏 ORM
class OrderRepository(Protocol):
def get(self, query: Select) -> Row | None:
...
失败表现：
- 应用层必须导入 ORM；
- 测试只能构造 ORM 查询；
- 领域对象变成 ORM 行对象；
- 数据库列名渗透到业务代码。
诊断问题：
如果换掉 ORM，端口的方法签名是否仍然有意义？
如果答案是否定的，端口定义在了适配器一侧，而不是业务一侧。
3. 用 mock 隐藏错误
repository.save.assert_called_once()
这只能说明调用发生了，不能说明：
- 保存后能否读回；
- 事务是否提交；
- 金额精度是否丢失；
- 状态是否正确映射；
- 重试是否产生重复数据。
mock 适合隔离协作关系，不适合替代所有真实边界。
4. 为了可替换而过度抽象
IOrderFactory
AbstractOrderFactory
DefaultOrderFactory
OrderFactoryProvider
OrderFactoryProviderRegistry
如果这些类没有隔离真实变化，只会增加调用路径和理解成本。
一个实用判断是：
只有当收益为正时，抽象才值得存在。这里的“成本”包括接口维护、测试数量、类型复杂度和调试难度，而不只是代码行数。
十五、与 Web API 工程和生产交付的连接
模块边界不是孤立的，它会直接影响 API 和交付。
API 契约
接口层可以独立演进：
API v1 请求
-> v1 DTO
-> 共享应用命令
-> 领域对象
API v2 请求
-> v2 DTO
-> 同一应用命令
-> 领域对象
只要业务语义没有变化，API 字段重命名、分页格式变化或错误响应格式变化不必修改领域层。
错误模型
领域错误、适配器错误和接口错误分层：
DomainError
-> 应用层保留语义
-> HTTP 层映射状态码和错误代码
如果领域层直接抛出 HTTPException，CLI 和消息消费者就会被迫理解 HTTP。
进程模型和资源生命周期
数据库连接池、HTTP 客户端、消息消费者应在进程启动时创建，在进程退出时关闭。组合根是管理这些资源的自然位置。
进程模型变化时：
单进程开发
-> 多 worker 生产
内存适配器中的状态不会自动跨 worker 共享。因此，限流计数、幂等记录和任务队列不能依赖单个 Python 进程内的全局变量。
灰度和回滚
可替换适配器也可以用于发布策略：
应用服务
├── 旧支付适配器
└── 新支付适配器
但选择逻辑应位于组合根或路由策略中，而不是散落在领域实体内。灰度时需要同时观察：
- 适配器错误率；
- 超时和重试次数；
- 业务成功率；
- 数据一致性；
- 新旧实现的结果差异。
如果新适配器写入了旧版本无法读取的数据，应用回滚仍可能失败。因此，适配器替换必须和数据契约、迁移兼容性一起设计。
十六、一套可执行的边界检查
提交代码前，可以对每个模块回答以下问题。
关于模块边界
- 模块对外暴露的是业务能力，还是数据库/框架对象？
- 调用者是否需要知道实现细节？
- 模块是否只有一个主要变化原因？
- 错误是否表达了调用者真正需要处理的语义？
关于依赖方向
- 领域层是否导入 Web、ORM、SDK 或环境变量？
- 是否存在两个包互相导入？
- 是否由组合根决定具体适配器？
- 替换数据库时，领域测试是否仍可运行？
关于领域层
- 不变量是否集中在值对象、实体或领域服务中？
- 外部入口是否都经过同一套状态转换规则？
- 金额、时间、ID 等关键输入是否有明确类型和边界？
- 领域对象是否被迫返回 HTTP 或 ORM 类型？
关于适配器
- 适配器是否实现了内层真正需要的端口？
- 适配器是否负责异常转换、资源管理和数据映射？
- 是否有契约测试？
- 是否明确事务、并发、重试和幂等语义？
最后可以用依赖检查工具或静态分析检查包之间的反向依赖，但工具只能发现部分结构问题。真正重要的是让依赖图能够被工程师解释：
接口层负责协议
应用层负责用例
领域层负责不变量
适配器负责外部系统
组合根负责选择和组装
当这五种责任边界稳定后，Python 项目的复杂度不会消失，但复杂度会被放置在可定位、可测试、可替换的位置。 Python 3.14 的模块、包和导入机制提供了组织代码的基础；架构则决定这些基础如何形成一张不会反向污染业务规则的依赖图。(docs.python.org)
系列导航与关联阅读
- 系列入口：Python 完整学习路线：从语言模型、并发到 Web、数据、AI 与生产交付
- 上一篇：Python 安全工程：输入、注入、反序列化、依赖、Secret 和沙箱
- 下一篇：FastAPI 完整基础：路由、依赖注入、校验、异步和生命周期
- 延伸：Python Web API 工程：契约、错误、分页、幂等、限流和版本
- 延伸：Python 生产交付：进程模型、容量、配置、迁移、灰度和回滚
官方资料
本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。
评论
0 条讨论