Python后端中间件专题01：中间件不是“多装几个容器”——请求之外的职责与故障地图
周一早上，客服群里同时出现三句话：“工单提交了，但没收到通知”、“搜索不到刚创建的单”、“Redis 超时了”。如果值班者把这三件事统称为“工单系统挂了”，他很可能做出错误恢复：因为搜索缺文档而重建业务行，因为缓存不可用而拒绝所有详情读取，或者把通知失败误报成创建失败。
第一课不启动任何中间件。我们先学会画故障地图：把 PostgreSQL 事实、Redis 加速副本、Elasticsearch 投影和异步通知效果分开，并为 Redis、搜索、通知故障各写出不篡改“工单已经提交”的用户契约。
阅读下面的图之前，确认你能读懂 Python Protocol/dataclass，知道 HTTP 失败会触发重试，也理解数据库事务整体提交或回滚。本篇不要求安装 Redis 或 Docker；按专题导读建立 Python 3.11 的 01–08 实验工作区，后面的 selector 都从 project/ 运行。
读图的第一问：事实有没有被碰到
TicketFlow 只承认 PostgreSQL 里的 tickets 行是工单事实。一次创建在同一数据库事务中写入工单、幂等收据和 Outbox 事件。只有这次提交成功，API 才有资格返回稳定工单 ID。
其他组件都是从属者：
- Redis 存了一份详情副本，可以过期或被整库清除；
- Elasticsearch 存了方便全文检索的投影，可以从 PostgreSQL 重建；
- RabbitMQ 运送待执行的工作，ACK 只表示消费者对消息做了处理决定，不是新的工单事实；
- 通知 sink 是异步副作用的可审计结果，它允许比 HTTP 响应晚到；
- 日志、指标和 trace 是证据，不参与业务裁决。
这种分类直接决定降级方式。Redis 宕机时，普通详情可以在有界限的前提下回源数据库；Elasticsearch 宕机时，搜索应清晰返回不可用，但创建和按 ID 读取不该被拖下水；通知延迟时，应查 Outbox、broker 和 worker 的交接，不能重新插入工单。
把这条判断锁在领域边界里
下面的 TicketService 看起来很薄，但它做了重要选择：所有业务操作只依赖 TicketStore 协议。它没有 RedisTicketStore、SearchTicketStore 或“哪个库先返回就信谁”的分支。数据库 repository 拥有事务；缓存与搜索只能包装读路径或消费已提交事件。
这份模块中有三个值得留意的约束。第一，每个读写方法都显式携带 tenant_id，租户隔离不是路由层的临时 if。第二，create 要求 idempotency_key，因为 HTTP 超时后客户端一定会重试。第三，IdempotencyConflict 是业务冲突，不应被吞掉并变成另一次创建。
用一条反例验证边界没有倒置
本课检查点创建一张 SQL 工单，然后在 Redis 写入一个只存在于缓存中的伪造文档。断言不只看 happy path：它还确认这个伪造 ID 不会凭空出现在 tickets 或 Outbox 中。缓存文档还要带由所有 API/Worker 共享部署密钥计算的 HMAC；结构完整但没有合法签名的 Redis 文档只能当作 miss，不能越过 PostgreSQL 的存在性裁决。
运行命令：python -m pytest tests/unit/test_architecture_contract.py::test_sql_repository_is_the_only_ticket_truth_and_adapters_are_non_authoritative -q
本地 Python 3.11 实际输出为：
. [100%]
1 passed
本地证据仍来自 SQLite、fake Redis 和 HTTP mock。可信远端 selector test_real_postgres_remains_ticket_truth_when_redis_state_disagrees 会在真实 PostgreSQL 中核对 ticket、幂等收据与 Outbox 各一行，保留一个可完整反序列化但无合法签名的 Redis-only 文档并要求 HTTP 仍为 404，再停 Redis 验证原工单仍可读。未执行远端 selector 前仍是 PENDING。
值班分诊时禁止的三种“修复”
- 把“搜索不到”解释为“工单没创建”，继而从 Elasticsearch 反写 PostgreSQL；这颠倒了事实源与投影。
- Redis 超时就让所有详情请求失败，而没有先判断该端点能否有界回源数据库。
- HTTP 已返回 201 后因通知延迟再次创建工单；正确恢复入口是原工单对应的 Outbox 事件。
本课练习
把故障地图变成分诊卡。
练习 EX-01-01 [code]
用一个可执行 Python 表格分别表达三种故障：创建已提交但通知延迟、Elasticsearch 宕机、Redis 超时。每项必须给出 truth_affected、user_contract 和 recovery，并用断言保证三种故障都不能把已提交工单改写为“未创建”。
练习 EX-01-02 [prose]
为 GET /live 和 GET /health 各写一句契约：哪个可以完全不访问外部依赖？PostgreSQL、Redis 或搜索故障时，哪个状态应决定 API 是否 ready？下一课会把这两条契约放到可测试的应用工厂旁边。
完成两题后，带着“先确认事实是否提交、再处理从属组件”的判断进入第 02 篇。它会逐题回答 EX-01-01/02，把健康契约变成可调用的 /live、依赖注入式 /health 和独立 selector，并建立 POST→GET→PATCH 的最小基线。
完整核心模块：工单用例与存储协议
"""Ticket use cases; persistence is delegated to a transaction-owning repository."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from typing import Optional, Protocol
class IdempotencyConflict(RuntimeError):
"""Raised when a retry key is reused for a different create request."""
@dataclass(frozen=True)
class Ticket:
id: str
tenant_id: str
actor_id: str
title: str
body: str
priority: str
status: str
version: int
created_at: datetime
updated_at: datetime
class TicketStore(Protocol):
def create_ticket(
self,
tenant_id: str,
actor_id: str,
title: str,
body: str,
priority: str,
idempotency_key: str,
) –> Ticket:
...
def get_ticket(self, tenant_id: str, ticket_id: str) –> Optional[Ticket]:
...
def update_ticket(
self,
tenant_id: str,
ticket_id: str,
title: Optional[str] = None,
body: Optional[str] = None,
priority: Optional[str] = None,
) –> Optional[Ticket]:
...
class TicketService:
"""Applies the application contract while keeping tenant-scoped storage explicit."""
def __init__(self, repository: TicketStore) –> None:
self._repository = repository
def create(
self,
tenant_id: str,
actor_id: str,
title: str,
body: str,
priority: str,
idempotency_key: str,
) –> Ticket:
return self._repository.create_ticket(
tenant_id, actor_id, title, body, priority, idempotency_key
)
def get(self, tenant_id: str, ticket_id: str) –> Optional[Ticket]:
return self._repository.get_ticket(tenant_id, ticket_id)
def update(
self,
tenant_id: str,
ticket_id: str,
title: Optional[str] = None,
body: Optional[str] = None,
priority: Optional[str] = None,
) –> Optional[Ticket]:
return self._repository.update_ticket(tenant_id, ticket_id, title, body, priority)
None,
) -> Optional[Ticket]:
return self._repository.update_ticket(tenant_id, ticket_id, title, body, priority)
