You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何使用Python处理JSON API请求超时并实现有效重试

原有代码的核心问题排查
  • 重试判断逻辑完全失效:代码里写的if 'Timeout' in x是检查字典x的顶层键名是否包含Timeout,而实际超时返回的字典结构是{'Message': 'xxx Execution Timeout Expired'},顶层键为Message,不存在带Timeout的键名,导致判断分支永远不会触发,整个重试逻辑等于空转。
  • 请求流程顺序颠倒:代码先执行了第一次请求拿到x,才进入while循环做判断,就算判断逻辑正确,第一次拿到的超时值也没有在循环入口做拦截,很容易漏处理异常返回。
  • 重试计数逻辑有漏洞:如果连续5次请求都返回超时,循环退出时x仍然是超时返回的内容,后续创建DataFrame的时候还是会触发值不匹配的报错。
  • 冗余代码无实际作用:代码里写的"{} json timed out. Trying again".format(id)只是生成了字符串对象,没有加print()执行,根本不会输出重试提示。
  • 未设置请求层超时:没有给requests.get加超时参数,遇到服务端无响应的情况会一直挂起,不会主动触发重试流程。
  • DataFrame创建方式容错性差:pd.DataFrame(index =[0]).from_dict(x)的写法在遇到标量字典时很容易触发ValueError报错,对返回内容的结构兼容度不足。
  • 变量名冲突:循环时用list作为迭代变量名,会覆盖Python内置的list类型,后续运行容易出现隐式bug。
优化实现思路

按照以下逻辑修改即可稳定解决偶发超时报错:

  • 把请求、结果校验、重试的全流程封装到循环内部,从入口就拦截异常返回
  • 修正超时判断逻辑:检查返回字典的Message字段值是否包含超时关键词,而非检查顶层键名
  • 给每个请求加网络层超时限制,避免请求永久挂起
  • 连续重试达到上限时做兜底处理:跳过异常id并记录失败日志,不硬写入DataFrame导致整个脚本中断
  • 统一用pd.DataFrame([x])的方式转换字典为DataFrame,从根源上避免标量值导致的创建报错
修正后参考代码
import requests
import time
import pandas as pd

all_records = []
# 替换为你实际的ID列表,避免用list作为变量名覆盖内置函数
for id in your_id_list:
    retry_count = 0
    max_retry = 5
    current_data = None
    # 带重试的请求流程
    while retry_count < max_retry:
        try:
            # 给请求加10秒超时,避免永久挂起
            resp = requests.get(
                f'你的目标API地址{id}',
                timeout=10
            )
            resp.raise_for_status() # 拦截4xx/5xx类的HTTP错误
            x = resp.json()
            # 正确判断是否为超时返回
            is_timeout = isinstance(x, dict) and 'Message' in x and 'Timeout' in str(x['Message'])
            if not is_timeout:
                current_data = x
                break
            # 命中超时返回,准备重试
            print(f"ID {id} 请求超时,第{retry_count+1}次重试")
            retry_count +=1
            time.sleep(1) # 间隔1秒再发请求,避免密集请求给服务端造成压力
        except Exception as e:
            print(f"ID {id} 请求异常:{str(e)},第{retry_count+1}次重试")
            retry_count +=1
            time.sleep(1)
    # 兜底处理:达到最大重试次数仍未拿到正常数据则跳过
    if current_data is None:
        print(f"ID {id} 连续重试{max_retry}次仍失败,已跳过")
        continue
    # 统一用列表套字典的方式转DataFrame,从根源避免标量值报错
    all_records.append(pd.DataFrame([current_data]))

# 合并所有记录生成最终的总DataFrame
final_df = pd.concat(all_records, ignore_index=True)

内容的提问来源于stack exchange,提问作者Booster

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.30 12:15:06