--- title: "POST 超时后怎么安全重试：用 Python 和 SQLite 给创建订单做幂等" date: 2026-11-12 slug: post-idempotency-key canonical: https://kernelranch.com/blog/2026/11/post-idempotency-key markdown_url: https://kernelranch.com/blog/2026/11/post-idempotency-key.md tags: ["python", "sqlite", "http", "idempotency"] ---订单已经提交，但客户端在收到 `201` 前超时。此时重发 `POST /orders` 有可能再建一笔。要让重试安全，客户端为一次业务操作生成一个 `Idempotency-Key`，服务端在同一个数据库事务里写入订单和这次请求的结果。重试用原键，就能取回第一次提交的响应。 ## 1. 先定接口契约 示例假设请求已通过鉴权，租户 ID 来自服务端的鉴权结果。这里只有 `POST /orders` 一个操作，所以幂等键的唯一范围是 `(tenant_id, key)`。如果多个操作共用一张表，把操作名也加进唯一键。 | 到达服务端的请求 | 处理结果 | |---|---| | 新键，参数合法 | 创建订单，保存并返回 `201` 和响应体 | | 同租户、同键、同参数 | 重放已保存的 `201` 和响应体 | | 同租户、同键、参数不同 | 返回 `409`，保留原订单和响应 | | 不同租户、相同键 | 各自处理，仍需独立鉴权 | 客户端在第一次发送前生成并保存键。超时、连接断开或按接口契约允许重试时，沿用原键和原参数。换了商品或数量，就开启一次新操作，生成新键。`409` 是这个示例定义的接口行为，接入已有 API 时以它自己的契约为准。 ## 2. 订单和重放记录一起提交 保存为 `idempotency_lab.py`，运行 `python idempotency_lab.py`。脚本只需要 Python 3.11 的标准库，会在临时目录创建数据库。`create_order()` 是 HTTP 路由可调用的业务函数，返回 `(status, body)`。 ```python import json import sqlite3 from concurrent.futures import ThreadPoolExecutor from contextlib import closing from pathlib import Path from tempfile import TemporaryDirectory from threading import Barrier def init_db(path): with closing(sqlite3.connect(path)) as db: db.executescript(""" CREATE TABLE orders ( id INTEGER PRIMARY KEY, tenant_id TEXT NOT NULL, sku TEXT NOT NULL, quantity INTEGER NOT NULL ); CREATE TABLE order_replays ( tenant_id TEXT NOT NULL, request_key TEXT NOT NULL, request_payload TEXT NOT NULL, status INTEGER NOT NULL, body TEXT NOT NULL, PRIMARY KEY (tenant_id, request_key) ); """) def create_order(path, tenant_id, key, sku, quantity): if not isinstance(key, str) or not 1 <= len(key) <= 128: return 400, '{"error":"invalid_key_or_sku"}' if not isinstance(sku, str) or not 1 <= len(sku) <= 100: return 400, '{"error":"invalid_key_or_sku"}' if type(quantity) is not int or not 1 <= quantity <= 100: return 400, '{"error":"invalid_quantity"}' # 从校验后的业务字段生成稳定表示，JSON 的空格与键顺序不参与比较。 payload = json.dumps({"sku": sku, "quantity": quantity}, sort_keys=True) with closing(sqlite3.connect(path, timeout=2.0)) as db, db: db.execute("BEGIN IMMEDIATE") saved = db.execute( """SELECT request_payload, status, body FROM order_replays WHERE tenant_id = ? AND request_key = ?""", (tenant_id, key), ).fetchone() if saved is not None: old_payload, status, body = saved if old_payload != payload: return 409, '{"error":"key_reused_with_different_payload"}' return status, body order = db.execute( "INSERT INTO orders (tenant_id, sku, quantity) VALUES (?, ?, ?)", (tenant_id, sku, quantity), ) status, body = 201, json.dumps({"order_id": order.lastrowid}) db.execute( """INSERT INTO order_replays (tenant_id, request_key, request_payload, status, body) VALUES (?, ?, ?, ?, ?)""", (tenant_id, key, payload, status, body), ) # 离开 with 时先提交事务，之后调用方才收到返回值。 return status, body def order_count(path): with closing(sqlite3.connect(path)) as db: return db.execute("SELECT count(*) FROM orders").fetchone()[0] def run_lab(path): init_db(path) first = create_order(path, "shop-a", "key-1", "book", 1) # 假装第一次响应在网络中丢失，客户端只收到这次重试的响应。 replay = create_order(path, "shop-a", "key-1", "book", 1) assert first == replay and replay[0] == 201 and order_count(path) == 1 print("retry:", *replay, "orders=1") conflict = create_order(path, "shop-a", "key-1", "book", 2) assert conflict[0] == 409 and order_count(path) == 1 print("changed payload:", *conflict, "orders=1") other = create_order(path, "shop-b", "key-1", "book", 1) assert other[0] == 201 and other != first and order_count(path) == 2 print("other tenant:", *other, "orders=2") start = Barrier(2, timeout=5) def submit(): start.wait() return create_order(path, "shop-a", "key-2", "book", 1) with ThreadPoolExecutor(max_workers=2) as pool: pending = [pool.submit(submit) for _ in range(2)] a, b = [future.result() for future in pending] assert a == b and a[0] == 201 and order_count(path) == 3 print("concurrent retry:", *a, "orders=3") with closing(sqlite3.connect(path)) as db, db: db.execute(""" CREATE TRIGGER fail_replay BEFORE INSERT ON order_replays WHEN NEW.request_key = 'key-3' BEGIN SELECT RAISE(ABORT, 'injected replay failure'); END; """) try: create_order(path, "shop-a", "key-3", "book", 1) except sqlite3.IntegrityError as error: assert str(error) == "injected replay failure" assert order_count(path) == 3 print("injected failure: rolled back, orders=3") else: raise AssertionError("expected the trigger to abort the insert") with closing(sqlite3.connect(path)) as db, db: db.execute("DROP TRIGGER fail_replay") recovered = create_order(path, "shop-a", "key-3", "book", 1) assert recovered[0] == 201 and order_count(path) == 4 print("retry after rollback:", *recovered, "orders=4") if __name__ == "__main__": with TemporaryDirectory() as tmp: run_lab(Path(tmp) / "orders.db") ``` `BEGIN IMMEDIATE` 在读重放记录之前就开启写事务。两个线程使用独立连接，SQLite 同一时间只允许一个写事务，因此第二个事务进入后能看到第一个事务已提交的结果。主键还在数据库层明确约束了键的作用域。触发器模拟第二次写入失败，异常离开 `with db` 时会连同已执行的订单写入一起回滚。 ## 3. 用订单数判断重试有没有多写 实测输出，Python 3.11.9 / SQLite 3.45.1： ```text retry: 201 {"order_id": 1} orders=1 changed payload: 409 {"error":"key_reused_with_different_payload"} orders=1 other tenant: 201 {"order_id": 2} orders=2 concurrent retry: 201 {"order_id": 3} orders=3 injected failure: rolled back, orders=3 retry after rollback: 201 {"order_id": 4} orders=4 ``` 并发实验同时放行两个线程，断言它们拿到相同的订单 ID，而且订单总数只增加一条。故障实验在写重放记录时抛错，订单总数保持在 3，清除故障后原键仍可正常完成创建。 ## 4. 接入真实路由时补齐三处边界 - **定义重放内容。** 如果 `201` 带有 `Location` 等业务响应头，把它一并保存或从已保存的订单 ID 稳定重建。使用校验后的业务字段比较请求参数。对含敏感字段的请求，考虑保存规范化内容的摘要，并约定规范化版本。 - **区分数据库繁忙和键冲突。** 示例的 SQLite 写事务会串行，写锁拿不到时可能出现 `SQLITE_BUSY`。HTTP 层可将可重试的繁忙映射为 `503`，客户端按退避策略带原键重试。`409` 留给同键不同参数。部署多个应用实例时，它们必须共享具有唯一约束的事务存储。每个实例各持一份本地数据库会各自创建订单。 - **明确保留期与副作用。** 保留期要覆盖客户端承诺的重试窗口。删除重放记录后，旧键可能再次创建订单。如果业务还有长期唯一性要求，给业务单号也加唯一约束。若订单创建还要触发扣库存服务或发送邮件，可在同一事务写入 outbox 事件，由后台投递，接收方再按事件 ID 去重。 客户端的超时设置可接着看 [Python HTTPX 用法](/blog/2026/11/python-httpx-sync-async.md)。 ## 小结 服务端判断一次重试能不能重放，需要同时看租户、操作、键和业务参数。用订单行数、响应体和事务失败后的状态验收，能覆盖客户端超时后最容易出问题的三条路径。 ## 数据来源与检索时间 检索时间 2026-11-12。 | 来源 | URL | 对应内容 | |---|---|---| | RFC 9110 §9.2.2 | https://www.rfc-editor.org/rfc/rfc9110.html#section-9.2.2 | 方法幂等性和重试的前提 | | SQLite 事务文档 | https://www.sqlite.org/lang_transaction.html | 单写事务、`BEGIN IMMEDIATE` 和 `SQLITE_BUSY` | | SQLite 触发器文档 | https://www.sqlite.org/lang_createtrigger.html#the_raise_function | 故障注入使用的 `RAISE(ABORT, ...)` | | Python 3.11 `sqlite3` 文档 | https://docs.python.org/3.11/library/sqlite3.html | 连接超时、参数绑定和连接上下文管理器 | ## 核验说明 - 在本地连续运行完整脚本 10 次，均通过同键重放、参数冲突、跨租户隔离、并发写入和事务回滚断言。 - 响应丢失通过丢弃第一次调用结果来模拟。实验覆盖数据库写入和重放逻辑，生产路由接入还需验证 HTTP 中间件、客户端超时与重试配置。