为什么Airflow请求Rest API时报错，第2000次请求出现KeyError: 'items'

KeyError: 'items'的本质是第2000次请求返回的JSON响应中不存在items字段，该问题仅在Airflow环境出现和运行环境的差异直接相关，按以下步骤排查解决：

1. 新增日志定位异常返回内容 #

在请求逻辑中增加状态码和响应报文的打印逻辑，确认第2000次请求时API的实际返回结果：

import requests
import logging

resp = requests.get(API_URL, params={"page": current_page, "page_size": PAGE_SIZE})
# 打印核心信息，过长报文截取前500字符即可
logging.info(f"当前页码：{current_page}，响应状态码：{resp.status_code}，响应内容：{resp.text[:500]}")
# 增加状态码校验，遇到4xx/5xx错误直接抛出异常
resp.raise_for_status()
json_data = resp.json()

如果返回状态码为429则说明触发了API限流，403说明授权信息过期，404则可能是页码超出了API实际的总页数。

2. 增加字段校验容错逻辑 #

不要硬编码取items字段，解析前先判断字段是否存在，异常情况触发重试或者终止逻辑：

if "items" not in json_data:
    logging.error(f"页码{current_page}返回无items字段，完整响应：{json_data}")
    # 可选择等待几秒后重试当前页，注意设置最大重试次数避免死循环
    # 如果确认已经请求到最后一页，可直接终止翻页循环

3. 适配Airflow运行环境差异 #

本地运行正常、Airflow运行报错的常见差异点：

• 请求速率：本地运行时通常存在隐式延迟，Airflow任务默认连续无间隔发起请求，很容易触发API的限流规则，可在每次请求之间加1~2秒的等待时间，或者配置指数退避的重试策略。
• 网络出口：Airflow工作节点通常使用集群共享出口IP，比你本地的独立IP更容易触发API的IP限流规则，可确认是否需要配置代理，或者联系API提供方将集群出口IP加入白名单。
• 参数配置：检查Airflow任务传入的请求参数是否和本地一致，例如页码起始值、每页返回数量，部分API限制单页最大返回量，参数超出限制时会返回异常报文。

4. 配置重试策略 #

针对API限流或偶发不稳定的场景，可增加自动重试逻辑，示例如下：

import time
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((KeyError, requests.exceptions.RequestException))
)
def get_single_page(page_num):
    resp = requests.get(API_URL, params={"page": page_num, "page_size": PAGE_SIZE})
    resp.raise_for_status()
    json_data = resp.json()
    if "items" not in json_data:
        raise KeyError("响应中无items字段")
    return pd.json_normalize(json_data["items"])

5. 优化翻页逻辑 #

不要硬编码总页码为5000，可根据API返回的total、has_next、next_page之类的字段判断是否还有下一页，避免请求超出实际总页数时返回异常报文。

内容的提问来源于stack exchange，提问作者HulkSmash