Python 实战：用 requests 批量调用 REST API 并清洗 JSON 数据

2026-07-17 Python

在平时的开发工作中，我们经常遇到这样的场景：需要把多个业务系统的数据汇总到一起，或者定时从第三方接口拉取数据做分析。如果靠手工在浏览器里点，效率低还容易出错。本文用 Python 的 requests 和 pandas，手把手教你写一个可复用的「批量调用 API → 解析 JSON → 清洗入库」小工具，跑一次就能完成以前几小时的重复劳动。

一、问题背景

假设你负责运营一个电商后台，每天要从订单系统、物流系统、会员系统三个接口分别拉取数据，合并成一张报表交给运营。三个接口返回的都是 JSON，但字段名不统一、有的字段缺失、有的时间格式乱七八糟。手工处理不仅慢，遇到接口限流还会失败。

二、核心思路

我们把任务拆成四步：

  1. 封装一个带重试和超时的请求函数，避免接口偶发失败导致整批任务中断；
  2. 用并发（线程池）批量调用，大幅缩短总耗时；
  3. 用 pandas 把 JSON 转成 DataFrame，统一字段名、处理缺失值；
  4. 导出为 Excel / CSV，或直接写回数据库。

三、完整代码实战

1. 封装健壮的请求函数

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def make_session():
    session = requests.Session()
    retry = Retry(total=3, backoff_factor=0.5,
                  status_forcelist=[429, 500, 502, 503, 504])
    session.mount('https://', HTTPAdapter(max_retries=retry))
    return session

def fetch_json(session, url, params=None, timeout=10):
    try:
        resp = session.get(url, params=params, timeout=timeout)
        resp.raise_for_status()
        return resp.json()
    except requests.RequestException as e:
        # 记录失败并返回 None，避免整批任务中断
        return None

2. 并发批量调用

from concurrent.futures import ThreadPoolExecutor

def batch_fetch(urls):
    session = make_session()
    with ThreadPoolExecutor(max_workers=5) as pool:
        results = list(pool.map(lambda u: fetch_json(session, u), urls))
    return [r for r in results if r]

3. 用 pandas 清洗数据

import pandas as pd

def clean(records):
    df = pd.DataFrame(records)
    # 统一字段名
    df = df.rename(columns={'order_id': '订单号', 'created': '下单时间'})
    # 处理缺失值与时间格式
    df['下单时间'] = pd.to_datetime(df['下单时间'], errors='coerce')
    df = df.dropna(subset=['订单号'])
    return df

urls = [f"https://api.example.com/orders?page={i}" for i in range(1, 6)]
data = batch_fetch(urls)
df = clean(data)
df.to_excel("orders_report.xlsx", index=False)

四、几个容易踩的坑

1. 接口限流：并发数别开太大，max_workers 控制在 5~10，并尊重 Rate-Limit 响应头，必要时加令牌桶限流。

2. 时间字段混乱：不同系统时间可能是时间戳、ISO 字符串或本地时间，统一用 pd.to_datetime(..., utc=True) 处理后再转换，避免时区错乱。

3. 编码问题：导出 Excel 或 CSV 时指定 encoding='utf-8-sig'，避免中文在 Excel 里变成乱码。

4. 大文件内存：数据量极大时分页读取，或用 chunksize 分块处理，别一次性 load 进内存导致 OOM。

五、总结

把「批量请求 + 并发 + pandas 清洗」这套组合拳固化成一个脚本，日常的数据汇总工作就能从几小时压缩到几分钟。这类自动化思路同样适用于日志分析、报表生成、接口监控等场景，是 Python 提升效率最实在的抓手之一。

关于我们

如果你在 Python 自动化、API 接口对接或 AI 集成方面遇到类似问题，尊云科技可以帮你解决。

服务范围：

  • Python 自动化脚本与数据处理
  • API 接口开发与第三方系统对接
  • AI 接口集成与大模型应用开发

联系方式：

  • 微信：yvsm316
  • QQ：316430983
标签： Python