Python 接口兼容示例：字段默认值、版本校验与幂等回放

原创

微订技术团队

修改于 2026-10-19 21:58:08

570

一个楼栋分配接口最初只接收楼栋编号和对象版本。后来需要增加备注字段，又不能要求所有旧客户端同时升级。字段能解析只是第一步：旧请求重试时，新版服务还需要判断它是不是同一次业务，并返回正确的历史结果。

本文用一个只依赖Python标准库的示例，讨论请求归一化、API版本、对象版本和幂等回放如何配合。代码使用单进程内存存储，目的在于复现接口行为，不提供重启后恢复或多进程一致性保证。

1. 三种标识不能混用

字段	它回答的问题	示例处理
api_version	这份请求按哪套字段规则解析	不支持的版本直接报错
expected_version	操作是否仍基于当前对象	首次执行时不匹配则冲突
request_id	这次业务操作以前是否处理过	同键同内容回放原结果

API版本改变不一定代表业务含义改变。例如v2新增remark，可约定旧客户端未传时按空字符串处理。只有当“未传”和“空字符串”在业务上含义一致时，才适合做这样的兼容。

对象版本则用于检查写入时机。两次不同操作即使使用同一API版本，也不能同时覆盖同一个旧对象版本。

2. 先归一化，再计算请求摘要

示例将v1的默认输入：

{"building_id": "B2", "expected_version": 1}

与v2的显式默认输入：

{"api_version": 2, "building_id": "B2", "expected_version": 1, "remark": ""}

归一化成相同的业务对象。字段排序和紧凑JSON编码只是稳定序列化的方法，真正决定是否等价的是字段默认值及业务语义。这里刻意不把api_version放入摘要，因为两个受支持适配器得到的是同一个操作；如果版本升级改变了语义，就不应沿用这个规则。

未知字段会报错，避免把拼错的字段静默丢弃。例如客户端把remark写成remarks，不能假装已经保存了备注。

3. 一份可运行的接口核心示例

保存下面的完整代码为demo.py，运行python demo.py即可。actor模拟服务端认证上下文，不是允许客户端填写的请求字段。示例只处理校区C1的一项分配对象。

import copy
import hashlib
import json
from threading import RLock

class Demo:
    def __init__(self):
        self.lock = RLock()
        self.version = 1
        self.saved = {}
        self.applied = 0

    def call(self, actor, request_id, body):
        # actor is server-side authentication context, not request JSON.
        if not actor["allowed"] or actor["campus"] != "C1":
            raise PermissionError("scope denied")
        if not isinstance(request_id, str) or not request_id.strip():
            raise ValueError("request_id required")
        api = body.get("api_version", 1)
        if type(api) is not int or api not in (1, 2):
            raise ValueError("unsupported API version")
        allowed = {"api_version", "building_id", "expected_version"}
        if api == 2:
            allowed.add("remark")
        if set(body) - allowed:
            raise ValueError("unknown field")
        building = body.get("building_id")
        expected = body.get("expected_version")
        remark = body.get("remark", "")
        if not isinstance(building, str) or not building.strip():
            raise ValueError("building_id required")
        if type(expected) is not int or expected < 1:
            raise ValueError("expected_version required")
        if not isinstance(remark, str):
            raise ValueError("remark must be string")
        semantic = {"building_id": building.strip(),
                    "expected_version": expected, "remark": remark}
        digest = hashlib.sha256(json.dumps(
            semantic, ensure_ascii=False, sort_keys=True,
            separators=(",", ":")
        ).encode("utf-8")).hexdigest()
        key = (actor["campus"], actor["id"], request_id)
        with self.lock:
            prior = self.saved.get(key)
            if prior:
                if prior["digest"] != digest:
                    raise ValueError("idempotency key conflict")
                result = copy.deepcopy(prior["result"])
            else:
                if expected != self.version:
                    raise ValueError("object version conflict")
                self.version += 1
                self.applied += 1
                result = dict(semantic, applied_version=self.version)
                self.saved[key] = {
                    "digest": digest, "result": copy.deepcopy(result)}
            return {"result": result, "current_version": self.version}

def expect_error(error, action):
    try:
        action()
    except error:
        return
    raise AssertionError("expected " + error.__name__)

def main():
    demo = Demo()
    actor = {"id": 7, "campus": "C1", "allowed": True}
    old = {"building_id": "B2", "expected_version": 1}
    first = demo.call(actor, "req-1", old)
    compatible = dict(old, api_version=2, remark="")
    assert demo.call(actor, "req-1", compatible) == first
    demo.call(actor, "req-2", {
        "api_version": 2, "building_id": "B3",
        "expected_version": 2, "remark": "east entrance"})
    replay = demo.call(actor, "req-1", old)
    assert replay["result"]["applied_version"] == 2
    assert replay["current_version"] == 3
    expect_error(ValueError, lambda: demo.call(
        actor, "req-1", dict(old, building_id="B9")))
    expect_error(ValueError, lambda: demo.call(
        actor, "req-1", dict(old, api_version=99)))
    expect_error(PermissionError, lambda: demo.call(
        dict(actor, allowed=False), "req-1", old))
    assert demo.applied == 2
    print("compatible replay: same saved result")
    print("saved/current version:", 2, 3)
    print("changed payload: rejected")
    print("unsupported API version: rejected")
    print("revoked permission: rejected")
    print("applied operations:", demo.applied)

if __name__ == "__main__":
    main()

幂等键由校区、操作者和请求编号组成。同一个请求编号在不同作用域下不应串用结果；真实接口仍需验证操作者是否有权访问具体分配对象，不能仅凭这三个字段就认定授权完成。

RLock让这个单进程例子中的“查历史、检查版本、写结果”不可被其他线程插入。它不是分布式锁，也不能代替数据库持久化事务。部署为多个进程或多个节点时，需要把业务变更和请求结果放进同一个持久化原子边界。

4. 为什么先查历史结果，再比较当前对象版本

首次req-1把对象从版本1更新到版本2，并保存applied_version=2。接着req-2合法地把对象更新到版本3。

此时旧客户端再次发送req-1，原请求里的expected_version仍为1。如果服务端先拿它和当前版本3比较，就会把一次成功操作的重试误判成新的写入冲突。

示例先在幂等记录中查找相同作用域和请求键：参数摘要一致就回放原result，不再执行第二次更新；同时用current_version告诉客户端当前对象已经到版本3。两个版本有不同含义，不能用最新状态覆盖历史结果。

返回前对保存结果做深拷贝，防止调用方修改响应对象后连带污染内存中的历史记录。

5. 本地实际验证的六种行为

本次在Windows环境使用Python 3.10.11运行，得到：

compatible replay: same saved result
saved/current version: 2 3
changed payload: rejected
unsupported API version: rejected
revoked permission: rejected
applied operations: 2

其中包含以下断言：v1请求和带默认备注的v2请求可回放同一结果；后续对象版本变化不会改变旧请求保存的applied_version；同键更换楼栋会被拒绝；未知API版本会被拒绝；权限撤销后即使有历史结果，也不能继续读到它；两次独立有效操作只产生两次业务变更。

权限检查放在读取幂等结果前面，避免“有缓存就跳过授权”。同样，API版本和字段解析也要先通过，不能让任意格式的请求只凭一个旧request_id取得结果。