如何在 API 调用中 Token 过期后从中断位置恢复 JSON 迭代-Python教程-PHP中文网

如何在 API 调用中 Token 过期后从中断位置恢复 JSON 迭代

本文介绍一种健壮的断点续传机制：当遍历 json 客户列表调用 api 因 token 过期中断时，自动记录最后失败的 clientid，并在下次运行时从此处继续，避免重复请求与数据遗漏。

在实际集成场景中（如对接 DocuSign、Salesforce 等需 Bearer Token 认证的 API），Token 通常具有较短有效期（如 1 小时），而批量处理数百个客户可能耗时更长。若每次 Token 失效都从头重跑，不仅浪费资源，还可能引发幂等性问题或限流风险。因此，实现「精准断点续传」——即严格从上一次中断的 ClientId 开始继续处理——是生产级脚本的关键能力。

✅ 核心设计思路

我们不依赖内存状态（因脚本退出后状态即丢失），而是采用轻量级持久化标记：

使用一个纯文本文件（如 last_failed.txt）存储最近一次 Token 过期时对应的 ClientId；
每次启动时检查该文件是否存在：
- 若不存在 → 全量遍历；
- 若存在 → 跳过此前所有 Client，从该 ID 对应项开始迭代；
成功完成全部请求后，自动清理该标记文件，表示任务已闭环。

? 实现要点解析

1. 扁平化嵌套结构，统一客户端流

原始 JSON 是双层嵌套（外层数组 → ClientList 数组），直接多层 for 循环易出错且难以跳转。推荐使用生成器抽象：

Remix V2 Meta Sessions Review

审查 Remix v2 代码中的 v1 形态 meta 导出（v2 中为破坏性变更），cookie 安全漏洞（httpOnly、secure、密钥轮换），认证门位于错误层级，以及其他

下载

def get_clients(data):
    """扁平化提取所有 Client 对象（支持任意层级嵌套）"""
    for outer in data:
        yield from outer.get("ClientList", [])

def get_clients_starting_from(data, client_id: str):
    """从指定 ClientId 开始生成客户端流（含自身）"""
    return itertools.dropwhile(
        lambda c: c["ClientId"] != client_id,
        get_clients(data)
    )

itertools.dropwhile 是关键：它会持续跳过元素，直到首次满足 c["ClientId"] == client_id，然后包含该元素及后续所有元素，完美匹配“从失败点继续”的语义。

2. 主流程：状态感知 + 容错执行

LAST_FAILED_PATH = pathlib.Path("last_failed.txt")

def main():
    with open("client_file.json") as f:
        data = json.load(f)

    # 决定起始位置
    if LAST_FAILED_PATH.exists():
        with open(LAST_FAILED_PATH) as f:
            start_id = f.read().strip()
        clients = get_clients_starting_from(data, start_id)
        print(f"▶ Resuming from ClientId: {start_id}")
    else:
        clients = get_clients(data)
        print("▶ Starting from beginning")

    # 执行请求循环
    for client in clients:
        result = call_api_with_retry(client)  # 建议加入重试逻辑（见下文）
        if result == "Expired":
            print(f"⚠ Token expired at ClientId: {client['ClientId']}")
            LAST_FAILED_PATH.write_text(client["ClientId"])
            break
    else:
        # 正常结束：删除断点标记
        if LAST_FAILED_PATH.exists():
            LAST_FAILED_PATH.unlink()
        print("✅ All clients processed successfully")

3. 生产增强建议（必加）

Token 自动刷新：在 call_api_with_retry() 中检测 401 Unauthorized 或 Session expired 错误，触发 refresh_token() 并重试当前请求（无需中断）；
请求重试机制：对网络超时、5xx 错误增加指数退避重试（如 tenacity 库）；
日志与监控：记录每个 Client 的耗时、状态码、错误详情到结构化日志（如 JSON Lines）；
并发控制：如需提速，可用 concurrent.futures.ThreadPoolExecutor 限制并发数（如 max_workers=5），但需注意 Token 共享安全；
幂等性保障：确保 API 支持幂等 Key（如 Idempotency-Key header），避免重复提交。