Python接口幂等性需综合请求来源、存储能力、业务语义与重试机制；通用方案是客户端传唯一idempotency_key，服务端用数据库唯一索引或缓存nx=True校验状态。
Python 接口幂等性不能靠“加个装饰器”一劳永逸，关键得看请求来源、存储能力、业务语义和失败重试机制是否匹配。
400-7654-355 idempotency_key 字段做服务端校验最通用客户端在每次请求中必须携带唯一、稳定、可预测的 idempotency_key（比如 UUIDv4 或业务侧生成的确定性哈希），服务端据此查库或缓存判断是否已处理过该键。
json）或 Header（如 X-Idempotency-Key）里显式传入，不能依赖 URL 或 query 参数（易被代理/CDN 缓存或日志泄露）CREATE UNIQUE INDEX idx_idempotency ON idempotency_records (key) WHERE status = 'success';，避免并发插入冲突pending、success、failed，不能只存成功结果——否则重试时无法知道上次是卡在中间态还是真失败了cache.set(key, result, timeout) + nx=True 防重复执行适合轻量级、无强事务要求的场景（如发通知、写日志），但要注意缓存穿透和一致性边界。
nx=True 是原子性保障核心，不加它可能两个请求同时通过 cache.get() 判断为空，然后都执行业务逻辑from django.core.cache import cachedef handle_payment(idempotency_key, amount): cache_key = f"idemp:{idempotency_key}"
尝试原子写入占位
if not cache.set(cache_key, "processing", timeout=600, nx=True): # 已存在，说明有请求正在处理或已完成 result = cache.get(cache_key) if result == "success": return {"status": "ok", "message": "already processed"} else: return {"status": "pending", "message": "in progress"} try:# 执行真实业务（扣款、发券等） do_actual_payment(amount) cache.set(cache_key, "success", timeout=600) return {"status": "ok", "message": "processed"} except Exception: cache.set(cache_key, "failed", timeout=600) raise
Flask + SQLAlchemy 场景下，
INSERT ... ON CONFLICT DO NOTHING是可靠底座PostgreSQL 的 upsert 能在数据库层拦截重复插入，比应用层锁更可靠，尤其适合需要落库即幂等的场景（如订单创建）。
# 执行真实业务（扣款、发券等）
do_actual_payment(amount)
cache.set(cache_key, "success", timeout=600)
return {"status": "ok", "message": "processed"}
except Exception:
cache.set(cache_key, "failed", timeout=600)
raiseidempotency_key 是表的唯一约束字段（或联合唯一），否则 ON CONFLICT 不生效RETURNING created_at，而不是用当前时间伪造服务端做了幂等，客户端却用指数退避反复 POST 同一个 body（含随机 nonce），那所有努力都白搭。
idempotency_key → 直到收到明确 success 响应才换 key409 Conflict 或 200 OK（带 X-Request-ID 和 X-Idempotent-Result: true），而不是一律 422 Unprocessable Entity
最难的不是写校验逻辑，而是定义清楚“什么才算一次相同操作”——转账接口里，金额相同但收款人不同算不算重复？部分退款和全额退款共用一个 key 是否合理？这些得和产品、支付网关、对账系统对齐，代码只是最后一环。
