API 幂等处理引擎（专业版）
输入 / 输出
输入：幂等键、当前时间；业务完成后的状态码与结果。
输出：三态动作（execute/replay/conflict）及首次缓存结果。
使用示例
from idempotency import IdempotencyStore
store = IdempotencyStore(ttl=86400)
key = request.headers["Idempotency-Key"] # 或 store.generate_key()
action, payload = store.acquire(key)
if action == "replay":
return payload["status"], payload["result"] # 200，重放
if action == "conflict":
return 409, {"error": "request in progress"}
try:
status, result = do_business() # execute
store.complete(key, status, result)
return status, result # 201
except Exception:
store.release(key) # 失败释放，允许重试
raise
引擎代码：idempotency.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""API 幂等性引擎：幂等键生成 + 首次执行/重复重放/进行中冲突 三态 + TTL 清理。"""
import uuid
import time
class IdempotencyStore:
def __init__(self, ttl=86400):
self.store = {} # key -> {status, result, created}
self.inflight = set()
self.ttl = ttl
def generate_key(self):
return str(uuid.uuid4())
def _purge(self, now=None):
now = now if now is not None else time.time()
for k in [k for k, v in self.store.items() if now - v["created"] > self.ttl]:
del self.store[k]
def acquire(self, key, now=None):
"""返回 (action, payload)：
execute = 首次，需执行业务；
replay = 重复请求，返回首次缓存；
conflict = 同键正在处理中（并发），返回 409。
"""
self._purge(now)
if key in self.store:
return ("replay", self.store[key])
if key in self.inflight:
return ("conflict", None)
self.inflight.add(key)
return ("execute", None)
def complete(self, key, status, result, now=None):
self.inflight.discard(key)
self.store[key] = {"status": status, "result": result,
"created": now if now is not None else time.time()}
def get_cached(self, key):
return self.store.get(key)
def release(self, key):
"""业务执行失败时释放 in-flight 锁，允许后续重试。"""
self.inflight.discard(key)
测试（3 正例 + 3 边界，实跑全过）
| 用例 | 断言 |
|---|---|
| 正例1 | 首次 acquire 返回 execute |
| 正例2 | complete 后重复 acquire 返回 replay，缓存状态码与结果一致 |
| 正例3 | generate_key 为合法 UUID4，1000 个全部唯一 |
| 边界1 | 同键 in-flight 未完成时并发 acquire 返回 conflict |
| 边界2 | TTL 过期后 acquire 重新返回 execute |
| 边界3 | 失败 release 后可重新 acquire 执行 |
硬性规则
- 幂等键绑定业务意图，重试复用、意图变更换新键。
- 只有业务成功才 complete 缓存；失败必须 release，不缓存失败结果。
- in-flight 同键并发一律 conflict，杜绝双写副作用。
- 本引擎为进程内示例实现；分布式部署需替换为 Redis 等共享存储（接口不变）。
