Python脚本调用API报KeyError:_embedded如何调试及容错处理

一、错误触发原因调试方法 #

• 新增响应上下文日志：每次请求完成后先打印HTTP状态码、当前请求页码、前500字符的原始响应内容，确认报错位置的返回结构是否符合预期。大概率是触发API限流、权限过期、分页参数超限后，接口返回了错误提示结构，而非正常的带_embedded字段的业务数据。
• 定位异常页码：循环中新增当前页码的打印/日志记录，报错后可直接单独请求异常页的URL，手动验证返回内容，排查是否为API本身的分页缺陷（比如只允许访问前25页）、单页数据结构异常。
• 异常时全量输出响应：捕获KeyError时直接将完整的响应JSON写入本地日志文件，不用猜测结构，直接确认缺少_embedded字段的根因。

二、异常容错实现方案 #

核心思路是分层捕获异常，对请求、解析、写入三个环节分别做容错，单页/单条异常不中断整体流程，同时留存日志方便后续补数据。

实现代码示例 #

import logging
import requests
import pandas as pd
import time
from sqlalchemy import create_engine

# 配置日志，同时输出到文件和控制台，方便回溯
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler("api_pull_runtime.log", encoding="utf-8"),
        logging.StreamHandler()
    ]
)

# 配置参数
API_BASE_URL = "替换为你的API接口地址"
TOTAL_PAGES = 400
PAGE_SIZE = 25
REQUEST_INTERVAL = 0.5 # 每页请求间隔，单位秒，避免触发限流
RETRY_TIMES = 3 # 单页请求失败重试次数
MYSQL_CONN = "mysql+pymysql://用户名:密码@地址:端口/库名?charset=utf8mb4"

all_valid_data = []
engine = create_engine(MYSQL_CONN)

for page_num in range(1, TOTAL_PAGES + 1):
    current_url = f"{API_BASE_URL}?page={page_num}&size={PAGE_SIZE}"
    logging.info(f"开始请求第{page_num}页，地址：{current_url}")
    page_skip_flag = False

    # 1. 单页请求重试逻辑
    resp = None
    for retry_idx in range(RETRY_TIMES):
        try:
            resp = requests.get(current_url, timeout=15)
            resp.raise_for_status() # 4xx/5xx状态码直接抛出异常
            break
        except Exception as e:
            logging.warning(f"第{page_num}页第{retry_idx+1}次请求失败：{str(e)}，等待10秒后重试")
            time.sleep(10)
    if not resp:
        logging.error(f"第{page_num}页重试{RETRY_TIMES}次全部失败，跳过该页")
        continue

    # 2. 响应结构校验与解析容错
    try:
        resp_json = resp.json()
        # 提前校验必要字段是否存在，避免KeyError
        if "_embedded" not in resp_json or "results" not in resp_json.get("_embedded", {}):
            logging.warning(f"第{page_num}页返回结构异常，响应内容：{str(resp_json)[:1000]}，跳过该页")
            continue
        page_records = resp_json["_embedded"]["results"]
    except Exception as e:
        logging.error(f"第{page_num}页响应解析失败：{str(e)}，跳过该页")
        continue

    # 3. 单条记录解析容错
    for record in page_records:
        try:
            # 你的单条数据处理逻辑，所有字段优先用get()取值，避免KeyError
            processed_record = {
                "record_id": record.get("id"),
                "title": record.get("title"),
                "create_time": record.get("createdAt"),
                # 补充你需要的其他字段
            }
            all_valid_data.append(processed_record)
        except Exception as e:
            logging.warning(f"第{page_num}页存在异常记录，已跳过，错误：{str(e)}，记录内容：{str(record)[:500]}")
            continue

    # 每页请求完成加间隔
    time.sleep(REQUEST_INTERVAL)

# 4. 数据导出与入库容错
if all_valid_data:
    df = pd.DataFrame(all_valid_data)
    # 导出CSV
    try:
        df.to_csv("api_full_data.csv", index=False, encoding="utf-8-sig")
        logging.info(f"CSV导出完成，共{len(df)}条有效数据")
    except Exception as e:
        logging.error(f"CSV导出失败：{str(e)}")
    # 写入MySQL
    try:
        df.to_sql(
            name="你的表名",
            con=engine,
            if_exists="append", # 按需修改为replace等
            index=False,
            chunksize=1000
        )
        logging.info(f"MySQL写入完成，共{len(df)}条数据")
    except Exception as e:
        logging.error(f"MySQL写入失败：{str(e)}")
else:
    logging.warning("本次未获取到任何有效数据")

关键优化点说明 #

• 所有字典取值优先用get()方法，字段不存在时返回None而非抛出KeyError
• 请求层加重试+间隔，大幅降低限流导致的结构异常概率
• 所有异常场景都留存日志，后续可以根据日志中的异常页码手动补拉数据，不会出现数据遗漏
• 单条记录异常只跳过当前条，不影响整页其他正常数据

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