两级接口依赖场景全解析:从字典列表到Excel导出的完整实战

累了不想盯屏幕？点耳机听全文吧～

特意安排了小朋友的声音陪你读，代码再枯燥，配上童声也变得温柔了

接口采集 · 数据对账 · 医保提取 · 第三方联动

一、场景还原：你在哪里见过这种结构？

日常开发中，你是否经常遇到这样的链路👇

一级接口（批量拉取）    ↓ list[dict]抽取指定字段    ↓二级接口（逐条调用）    ↓汇总所有返回数据    ↓导出 Excel

典型场景包括但不限于：

场景	一级接口	二级接口
医保数据提取	参保人员清单	个人就诊明细
业务对账	订单号列表	订单详情/流水
第三方爬虫	商品ID列表	商品详情/SKU
接口采集	机构编码列表	机构下人员信息

核心特征就一句话：

一级接口给你一个 list[dict]，你得从中拆字段，再去调二级接口，最后把所有结果拼回去。

二、新手必踩的4个坑（看看你中了几个）

❌ 坑1：循环里直接赋值，结果只剩最后一条

# ❌ 错误写法result = {}for item in data:    result = item  # 每次循环都覆盖！

最终结果：只有最后一次循环的 item 存活。

❌ 坑2：列表 append 位置不对，数据"消失"

# ❌ 错误写法all_data = []for item in data:    all_data = []  # 每次循环重新初始化！    all_data.append(item)

最终结果：每次循环清空列表，白忙活。

❌ 坑3：嵌套字典直接丢给 pandas，Excel 炸了

# ❌ 二级接口返回的嵌套结构{    "code": 0,    "data": {        "person": {"name": "张三", "age": 30},        "orders": [{"id": 1}, {"id": 2}]    }}

直接 pd.DataFrame(data) → 报错或一列全是 dict 对象。

❌ 坑4：多批次分页拉取，拼接后顺序乱了

分页拉取 + 二级接口调用 + 异步并发，最后发现：

数据重复 ✅
数据缺失 ✅
顺序不对应 ✅

三、正确姿势：完整原理拆解

✅ 核心思路一张图

┌─────────────────────────────────────────────┐│  一级接口返回：list[dict]                    ││  [{"id":1,"name":"A"}, {"id":2,"name":"B"}] │└──────────────┬──────────────────────────────┘               ↓ 遍历每条 dict┌─────────────────────────────────────────────┐│  抽取字段：id → 作为二级接口入参              ││  second_api(id)                             │└──────────────┬──────────────────────────────┘               ↓ 拿到二级返回┌─────────────────────────────────────────────┐│  合并策略：原 dict + 二级返回 dict           ││  {"id":1,"name":"A","second_data": {...}}   │└──────────────┬──────────────────────────────┘               ↓ 全部收集┌─────────────────────────────────────────────┐│  final_list = []                            ││  final_list.append(merged_dict)             │└──────────────┬──────────────────────────────┘               ↓ 扁平化 + 导出┌─────────────────────────────────────────────┐│  pd.DataFrame → Excel                       │└─────────────────────────────────────────────┘

四、可直接复制运行的成品代码 🚀

📦 环境准备

pip install requests pandas openpyxl

🧩 完整示例代码

import requestsimport pandas as pdimport timefrom typing import List, Dictclass TwoLevelApiCollector:    """两级接口数据采集器"""    def __init__(self):        self.session = requests.Session()        self.session.headers.update({            "Content-Type": "application/json"        })    def call_first_api(self) -> List[Dict]:        """一级接口：批量拉取基础数据"""        # 实际使用时替换为真实接口        url = "https://api.example.com/first"        resp = self.session.get(url, timeout=30)        resp.raise_for_status()        return resp.json().get("data", [])    def call_second_api(self, primary_id: str, **kwargs) -> Dict:        """二级接口：根据一级字段逐条调用"""        url = "https://api.example.com/second"        payload = {            "id": primary_id,            **kwargs        }        resp = self.session.post(url, json=payload, timeout=30)        resp.raise_for_status()        return resp.json().get("data", {})    def flatten_dict(self, d: Dict, parent_key: str = "", sep: str = "_") -> Dict:        """        嵌套字典扁平化 —— 这是转Excel的关键！        示例：        {"a": 1, "b": {"c": 2, "d": {"e": 3}}}        → {"a": 1, "b_c": 2, "b_d_e": 3}        """        items = []        for k, v in d.items():            new_key = f"{parent_key}{sep}{k}" if parent_key else k            if isinstance(v, dict):                items.extend(self.flatten_dict(v, new_key, sep).items())            elif isinstance(v, list):                # 列表处理方式：转为字符串，或展开为多列                # 这里选择转为字符串，避免DataFrame创建失败                items.append((new_key, str(v)))            else:                items.append((new_key, v))        return dict(items)    def collect_all(self) -> List[Dict]:        """完整采集流程"""        # ✅ 关键1：列表定义在循环外        final_result = []        # 一级接口        first_data = self.call_first_api()        print(f"一级接口返回 {len(first_data)} 条数据")        for idx, item in enumerate(first_data, 1):            # ✅ 关键2：深拷贝原字典，避免引用污染            merged = item.copy()            try:                # 从当前字典抽取字段                primary_id = item.get("id")                if not primary_id:                    print(f"第 {idx} 条缺少 id，跳过")                    continue                # 二级接口调用                second_data = self.call_second_api(                    primary_id=primary_id,                    extra_param=item.get("extra", "")                )                # ✅ 关键3：二级数据作为子字段挂上去                merged["second_response"] = second_data                # ✅ 关键4：扁平化处理（转Excel前）                flattened = self.flatten_dict(merged)                final_result.append(flattened)                print(f"第 {idx}/{len(first_data)} 条处理完成")                # 接口限速保护                time.sleep(0.1)            except Exception as e:                print(f"第 {idx} 条处理失败: {e}")                # 失败也要保留原始数据                merged["error"] = str(e)                final_result.append(self.flatten_dict(merged))                continue        print(f"✅ 采集完成，共 {len(final_result)} 条")        return final_result    def export_to_excel(self, data: List[Dict], filename: str = "output.xlsx"):        """导出Excel"""        if not data:            print("⚠️ 无数据可导出")            return        df = pd.DataFrame(data)        # ✅ 关键5：列顺序整理（可选）        # 把常用字段放前面        priority_cols = ["id", "name", "error"]        other_cols = [c for c in df.columns if c not in priority_cols]        df = df[priority_cols + sorted(other_cols)]        df.to_excel(filename, index=False, engine="openpyxl")        print(f"📊 已导出: {filename} ({len(df)} 行 × {len(df.columns)} 列)")    def run(self, output_file: str = "result.xlsx"):        """一键执行"""        data = self.collect_all()        self.export_to_excel(data, output_file)        return dataif __name__ == "__main__":    collector = TwoLevelApiCollector()    collector.run("医保数据_采集结果.xlsx")

五、核心转换逻辑详解（重点！）

🎯 flatten_dict：为什么必须做这一步？

原始结构	DataFrame 行为	扁平化后
`{"a": {"b": 1}}`	一列，值是 dict 对象	`{"a_b": 1}`
`{"x": [1,2,3]}`	一列，值是 list 对象	`{"x": "[1, 2, 3]"}`
`{"level1": {"level2": {"level3": "val"}}}`	多层嵌套，Excel 直接裂开	`{"level1_level2_level3": "val"}`

一句话总结：

pandas 只认"一层"的字典，嵌套的必须拍扁。

🎯 合并策略的三种选择

场景	策略	适用情况
一对一	`original.update(second)`	二级返回字段少，无冲突
一对多	`original["details"] = second_list`	二级返回列表，如订单明细
复杂嵌套	`original["second"] = second` + flatten	医保、风控等复杂业务

本文示例用的是第三种，最通用、最稳妥。

六、进阶优化建议 ⚡

1️⃣ 并发提速（适合大批量）

from concurrent.futures import ThreadPoolExecutor, as_completeddef batch_collect_concurrent(self, max_workers=5):    final_result = []    first_data = self.call_first_api()    with ThreadPoolExecutor(max_workers=max_workers) as executor:        futures = {            executor.submit(self.process_one, item): item             for item in first_data        }        for future in as_completed(futures):            result = future.result()            if result:                final_result.append(result)    return final_result

⚠️ 注意：并发时要加锁或线程安全的容器，同时注意接口限流！

2️⃣ 断点续传（防止中途崩溃）

import jsondef save_checkpoint(data, checkpoint_file="checkpoint.json"):    with open(checkpoint_file, "w", encoding="utf-8") as f:        json.dump(data, f, ensure_ascii=False, indent=2)def load_checkpoint(checkpoint_file="checkpoint.json"):    try:        with open(checkpoint_file, "r", encoding="utf-8") as f:            return json.load(f)    except FileNotFoundError:        return []

3️⃣ 失败重试机制

from tenacity import retry, stop_after_attempt, wait_exponential@retry(    stop=stop_after_attempt(3),    wait=wait_exponential(multiplier=1, min=1, max=10))def call_second_api_with_retry(self, primary_id):    return self.call_second_api(primary_id)

七、总结 Checklist ✅

下次写两级接口采集时，对照这张表自查：

[ ] 结果列表定义在循环外部
[ ] 使用 .copy() 避免字典引用污染
[ ] 嵌套字典在转 DataFrame 前做了扁平化
[ ] 接口调用加了异常捕获和超时控制
[ ] 失败数据单独标记，不影响整体流程
[ ] 大批量场景考虑了限速或并发控制
[ ] 导出前整理了列顺序，方便业务方查看

💬 写在最后

两级接口依赖是数据采集中最基础也最容易出错的环节。看似简单的 list[dict] 流转，背后藏着引用传递、数据结构设计、异常处理等多个知识点。

这篇文章的代码你可以直接复制改改接口地址就能跑，但更重要的是理解每一步背后的原理。

踩过的坑，别再踩第二次。 🫡

📌 如果这篇对你有帮助，欢迎点赞 + 在看 + 转发

🏷️ 文章标签

#接口采集#数据对账#Python#Pandas#Excel导出#医保数据#爬虫实战#后端开发#数据处理#避坑指南

推荐阅读：

从Excel重复劳作到自动出Word报告：Python办公自动化完整实战路线
merge合并两张表冗余列太多？两种方法精准只保留需要的字段
Pandas报错踩坑实录：isin多条件筛选传参错误，90%新手都会写错！

觉得不错的话，点个关注不迷路 👋