# 一本免费书，让你离场后集成代码依然好改

> 两位前 MADE.com 工程师合著，讲清一个 Flask 路由如何长成“大泥球”，以及怎样避免。

原文: https://fdetimes.net/zh/books-courses/architecture-patterns-with-python-review/

一整本软件架构书就放在网上，免费可读。cosmicpython.com 以 Creative Commons 许可提供《Architecture Patterns with Python》全文，并附有 O'Reilly Learning 的链接。

作者是 Harry Percival 和 Bob Gregory。两人在 MADE.com 共事期间写下这本书。MADE.com 是一家在线销售家具的欧洲电商公司。

他们设想的读者，是那些被越写越乱的代码折磨过的 Python 开发者。前言说得很直白：不需要懂 DDD，也不需要了解经典架构模式。

对前线部署工程师（FDE）来说，这本书正好戳中这份工作最痛的地方。写出能跑的集成代码不难，难的是你离开项目之后，这些代码还能继续改。作者在前言里点明了这个问题：项目越大，最直观的写法越帮不了你管理复杂度。

## 泥潭从第一个接口开始

引言用一句很直接的观察引出问题：软件陷入混乱太常见了，工程师们甚至给它起了名字，叫“大泥球”（Big Ball of Mud）反模式。

设想一次典型的部署。你写了一个 Flask 路由：接收请求，读数据库，把数据推送到客户的 CRM，然后提交。演示跑通了。

```python
# app.py — 改造前：所有逻辑都写在路由里
@app.route("/orders/<order_id>/sync", methods=["POST"])
def sync(order_id):
    order = db.session.get(Order, order_id)
    crm.push(order.to_dict())
    order.synced = True
    db.session.commit()
    return {"ok": True}
```

到了第三周，客户想要一个夜间任务，把所有数据重新同步一遍，于是你把中间那四行复制进一个脚本。现在逻辑有了两份副本，从第一次修改开始，它们就会各走各的路。

## 思路一：Web 只是入口之一

第 4 章针对的正是这种情况，给出的答案是服务层。按作者的说法，所有编排逻辑都放在用例层，业务逻辑则留在领域层。这里的编排指的是：从 repository 取数据、根据数据库状态校验输入、处理错误，一切正常时提交。

照这个思路改写示例，逻辑挪进一个独立函数，路由只剩几行：

```python
# services.py — 一个用例，所有入口都调用它
def sync_order(order_id, session, crm):
    order = session.get(Order, order_id)
    crm.push(order.to_dict())
    order.synced = True
    session.commit()

# app.py — 路由只做 Web 层的事
@app.route("/orders/<order_id>/sync", methods=["POST"])
def sync(order_id):
    services.sync_order(order_id, db.session, crm_client)
    return {"ok": True}
```

夜间任务遍历订单，逐个调用 `sync_order`；测试则直接用假的 session 和假的 CRM 调用它。逻辑只在一处，接手代码的人也只需要读这一处。

**要点:** 好的交接代码，是别人不用给你打电话就能测试的代码。

到客户现场先做的一件事：在写第二个路由之前，把第一个路由的逻辑抽成用例。

## 思路二：Unit of Work 把操作归为一个整体

第 6 章引入 Unit of Work，这是书中对原子操作这一概念的封装：把必须同时成功或同时失败的操作归为一个整体。不显式调用 commit，就什么都不会保存。借助 Python 的上下文管理器，未提交的工作会在请求结束时自动回滚。

但再看一眼 `sync_order`：CRM 在提交之前就收到了数据。如果提交失败，CRM 里已经有了这笔订单，数据库却仍记录为未同步。回滚只能撤销数据库里的内容，撤不回已经发出去的调用。

把顺序反过来，先提交再调用 CRM，也救不了你。AWS Prescriptive Guidance 提醒：如果服务在提交事务后立刻崩溃，发给另一个系统的通知可能永远发不出去。

Confluent 把这称为双写问题：数据库更新成功，外部写入却失败，系统由此处于不一致状态。

## 思路三：对外调用必须经过 outbox

AWS 和 Confluent 给出的解法都是事务性 outbox。在保存订单的同一个事务里，把“要调用 CRM”这个意图写进 outbox 表，再由一个独立的 worker 负责调用、失败时重试、发送成功后标记完成。放进书中的 Unit of Work 后，`sync_order` 变成这样：

```python
# services.py — 改造后：订单和 outbox 是一个整体
def sync_order(order_id, uow):
    with uow:
        order = uow.orders.get(order_id)
        order.sync_status = "pending"
        uow.outbox.add(event="order_sync", order_id=order_id,
                       idempotency_key=f"order-{order_id}")
        uow.commit()  # 要么都保存，要么都不保存

# worker.py — 独立运行，调用 CRM 并负责重试
def flush_outbox(uow, crm):
    with uow:
        for msg in uow.outbox.pending():
            order = uow.orders.get(msg.order_id)
            crm.push(order.to_dict(), idempotency_key=msg.idempotency_key)
            order.sync_status = "synced"
            msg.mark_done()
            uow.commit()
```

订单和 outbox 记录一起提交、一起回滚，因此不会再出现订单已保存、调用 CRM 的意图却丢失的情况。如果 CRM 出错，消息留在 outbox 里，等 worker 下次运行时重试。

代价是消息至少投递一次，也就是说 CRM 可能收到重复消息。AWS 建议让接收方具备幂等性，记录已处理过的消息，`idempotency_key` 承担的就是这个角色。如果客户的 API 不接受幂等键，worker 就得在重发前自己检查。

一个自测小练习：在 `flush_outbox` 中，假设 `crm.push` 成功了，但 worker 恰好在 `uow.commit()` 之前崩溃。此时订单和消息分别处于什么状态？下一次运行会做什么？先自己回答，再往下读。

答案：因为没有任何提交，Unit of Work 会回滚，消息仍是待处理，订单仍是未同步。下一次运行时，worker 用同一个 `idempotency_key` 重发，正是这个键阻止 CRM 创建第二笔订单。两半都答对了，你就明白为什么 outbox 和幂等性必须同时出现。

到客户现场先做的一件事：列出代码中所有既写数据库又调用外部系统的地方，逐一追问：第二步失败会怎样？

## 常见误区：还没痛就先加层

读完一本模式类的书，最大的诱惑是从第一个接口起就把每一层都搭出来。对 FDE 而言，每多一层，客户团队在你离开后就要多理解一样东西。

如果集成只涉及数据库，一个简单的提交函数依然够用。加层要有具体的痛点，比如逻辑被复制到了第二个任务里，或者某个对外调用可能丢失，而不是因为书里有这一章。

## 怎么读才不会被淹没

Giuliano Pertile 在 2022 年的一篇书评中仍然推荐这本书，但提醒书中给读者的练习相当难。这个提醒很中肯，也正是应该结合真实项目来读的原因。

比较合理的顺序：先读引言，理解代码为什么会滑向泥潭；再读服务层一章，然后是 Unit of Work。其余部分留到项目真正需要时再看。

从职业角度看，这是一项很容易展示的能力。一个个人仓库，路由足够薄，服务层清晰，带幂等键的 outbox，再加上一套不依赖真实数据库就能跑的测试，比简历上一句“整洁架构”更有说服力。FDE 面试中如果被问到你做过的集成，就讲讲外部系统中途失败时你是怎么处理的。

大多数客户永远不会读你的代码。但六个月后接手的工程师会读，而这本书正是为这个人写的。

**本周动手试试:**

- 在 cosmicpython.com 上读完引言和服务层一章，然后挑一个承载了过多逻辑的 Flask 或 FastAPI 路由
- 把这个路由的逻辑抽成一个用例函数，并写一个不启动 Web 服务器、直接调用该函数的测试
- 在代码库里找一处既提交数据库又调用外部系统的地方，在 README 中写明任一步失败时系统会怎样表现

## 信源

- [Cosmic Python](https://www.cosmicpython.com/)

- [Preface – Architecture Patterns with Python](https://www.cosmicpython.com/book/preface.html)

- [Introduction – Architecture Patterns with Python](https://www.cosmicpython.com/book/introduction.html)

- [Unit of Work Pattern](https://www.cosmicpython.com/book/chapter_06_uow.html)

- [Our First Use Case: Flask API and Service Layer](https://www.cosmicpython.com/book/chapter_04_service_layer.html)

- [Transactional outbox pattern – AWS Prescriptive Guidance](https://docs.aws.amazon.com/prescriptive-guidance/latest/cloud-design-patterns/transactional-outbox.html)

- [Dual write problem – Confluent](https://www.confluent.io/blog/dual-write-problem/)

- [Book Review: Architecture Patterns with Python](https://www.giulianopertile.com/blog/book-review-architecture-patterns-with-python/)