API 幂等防重判定引擎（专业版）

输入

make_idempotency_key(biz_prefix, request_body="", namespace="")；classify_idempotent(stored, same_payload=True)，stored 为 None 或 {{"state":"PROCESSING"/"COMPLETED"}}

输出

幂等键字符串；(动作, HTTP码, 是否执行副作用bool)

硬性规则

1. 同体同键、异体异键(体取SHA256前16位)。
2. None→EXECUTE_FIRST(200,True)；PROCESSING→IN_FLIGHT(409,False)；COMPLETED同体→REPLAY_CACHED(200,False)；同键异体→CONFLICT(422,False)。

示例

make_idempotency_key("order",'{"id":1}',"u1") == make_idempotency_key("order",'{"id":1}',"u1")
classify_idempotent({{"state":"COMPLETED"}}, True) → ("REPLAY_CACHED",200,False)
classify_idempotent({{"state":"COMPLETED"}}, False) → ("CONFLICT",422,False)

Python 引擎（可独立运行）

# -*- coding: utf-8 -*-
import hashlib

def make_idempotency_key(biz_prefix, request_body="", namespace=""):
    """由业务前缀+命名空间+请求体摘要生成确定性幂等键；同体同键、异体异键。"""
    body_hash = hashlib.sha256((request_body or "").encode("utf-8")).hexdigest()[:16]
    parts = [biz_prefix, namespace, body_hash]
    return "_".join([p for p in parts if p])

def classify_idempotent(stored, same_payload=True):
    """依据服务端该幂等键的存储记录判定动作，保证同一业务请求副作用只执行一次。
    stored=None：首次；dict: {"state":"PROCESSING"/"COMPLETED"}。
    same_payload：本次请求体与首次是否一致。
    返回 (动作, HTTP状态码, 是否执行业务副作用)。"""
    if stored is None:
        return ("EXECUTE_FIRST", 200, True)        # 首次：落 PROCESSING 并执行
    if not same_payload:
        return ("CONFLICT", 422, False)            # 同键不同体：拒绝，防键复用
    state = stored.get("state")
    if state == "PROCESSING":
        return ("IN_FLIGHT", 409, False)           # 并发重复：进行中，不重复执行
    if state == "COMPLETED":
        return ("REPLAY_CACHED", 200, False)       # 已完成：回放原结果，不重复扣款/下单
    return ("EXECUTE_FIRST", 200, True)