POST 超时后怎么安全重试：用 Python 和 SQLite 给创建订单做幂等

订单已经提交，但客户端在收到 201 前超时。此时重发 POST /orders 有可能再建一笔。要让重试安全，客户端为一次业务操作生成一个 Idempotency-Key，服务端在同一个数据库事务里写入订单和这次请求的结果。重试用原键，就能取回第一次提交的响应。

1. 先定接口契约

示例假设请求已通过鉴权，租户 ID 来自服务端的鉴权结果。这里只有 POST /orders 一个操作，所以幂等键的唯一范围是 (tenant_id, key)。如果多个操作共用一张表，把操作名也加进唯一键。

客户端在第一次发送前生成并保存键。超时、连接断开或按接口契约允许重试时，沿用原键和原参数。换了商品或数量，就开启一次新操作，生成新键。409 是这个示例定义的接口行为，接入已有 API 时以它自己的契约为准。

2. 订单和重放记录一起提交

保存为 idempotency_lab.py，运行 python idempotency_lab.py。脚本只需要 Python 3.11 的标准库，会在临时目录创建数据库。create_order() 是 HTTP 路由可调用的业务函数，返回 (status, body)。

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：

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 用法。

小结

服务端判断一次重试能不能重放，需要同时看租户、操作、键和业务参数。用订单行数、响应体和事务失败后的状态验收，能覆盖客户端超时后最容易出问题的三条路径。

数据来源与检索时间

检索时间 2026-11-12。

核验说明

• 在本地连续运行完整脚本 10 次，均通过同键重放、参数冲突、跨租户隔离、并发写入和事务回滚断言。
• 响应丢失通过丢弃第一次调用结果来模拟。实验覆盖数据库写入和重放逻辑，生产路由接入还需验证 HTTP 中间件、客户端超时与重试配置。