如何用脚本批量调用API接口？

wen 实用脚本 2026-07-24 4

本文目录导读：

如何用脚本批量调用API接口？

📖 目录导读
为什么需要批量调用API？——场景与痛点
核心准备：API接口文档与认证方式
脚本语言选择：Python、Bash、Node.js对比
手把手实战：Python脚本批量调用REST API
进阶技巧：异步调用与分布式任务
常见问题与问答（FAQ）
让批量调用成为生产力

如何用脚本批量调用API接口（实战指南）

📖 目录导读

为什么需要批量调用API？——场景与痛点
核心准备：API接口文档与认证方式
脚本语言选择：Python、Bash、Node.js对比
手把手实战：Python脚本批量调用REST API
- 1 单次调用基础模板
- 2 并发控制与速率限制
- 3 错误重试与日志记录
进阶技巧：异步调用与分布式任务
常见问题与问答（FAQ）
让批量调用成为生产力

为什么需要批量调用API？——场景与痛点

在当今的软件开发和运维中,调用外部API（如天气数据、支付网关、AI模型接口）是家常便饭，但当你的需求从“每天查一次”变成“每分钟查询1000条用户信息”时，手动修改参数或复制粘贴代码就成了噩梦。

典型场景：

批量更新数据库中的用户字段（如根据手机号查询归属地）
从第三方平台下载大量报告或文件（如GitHub Actions批量获取仓库Issue）
测试新服务的压力阈值（性能测试）
数据清洗后同步至多个系统

痛点：

手动反复调用容易出错,且耗时巨大
单线程调用速度慢,无法利用多核CPU
缺乏异常处理,一个错误卡住整个流程
未考虑调用频率限制,被接口封禁

核心准备：API接口文档与认证方式

在写脚本之前,你必须清楚三件事：

接口地址（Endpoint）：https://api.example.com/v1/users/{id}
认证方式：最常见的是 Bearer Token（Bearer Token）、API Key 或 OAuth 2.0，Token通常放在请求头 Authorization: Bearer xxx 中。
速率限制（Rate Limit）：文档中会写明“每分钟最多X次请求”，例如免费版OpenAI API限制为每分钟3次（RPM）。

建议： 在本地用Postman或curl先测试单次调用成功后再写脚本。

curl -H "Authorization: Bearer 你的Token" "https://api.example.com/v1/users/123"

脚本语言选择：Python、Bash、Node.js对比

语言	适合场景	优点	缺点
Python	数据密集型、需要复杂逻辑	库丰富（requests, asyncio）	学习曲线平，但并发稍慢（GIL）
Bash	快速试错、Linux管道操作	零安装，极轻量	处理JSON麻烦，错误控制弱
Node.js	高并发I/O密集型	事件驱动，天然异步	回调地狱（可用async/await解决）

推荐： 新手从 Python 开始，因为它有最友好的报错提示和成熟的第三方库。

手把手实战：Python脚本批量调用REST API

1 单次调用基础模板

import requests
import json
# 配置
API_BASE = "https://api.example.com/v1"
TOKEN = "your_token_here"
HEADERS = {"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"}
# 单条数据请求
def fetch_user(user_id):
    url = f"{API_BASE}/users/{user_id}"
    response = requests.get(url, headers=HEADERS)
    if response.status_code == 200:
        return response.json()
    else:
        print(f"用户{user_id}请求失败，状态码：{response.status_code}")
        return None

2 并发控制与速率限制

最基础的做法是用 time.sleep() 控制速率，但这太慢，更优雅的方法是使用 concurrent.futures 线程池（或 asyncio 异步协程），以下是线程池版：

from concurrent.futures import ThreadPoolExecutor, as_completed
import time
user_ids = list(range(1, 1001))  # 假设要查询1000个用户
results = []
# 限制并发数，避免触发速率限制
MAX_WORKERS = 10  
INTERVAL = 1.0 / 60  # 每分钟60次请求，即每1秒1次
def wrapper(user_id):
    time.sleep(INTERVAL)  # 简单限流
    return fetch_user(user_id)
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
    futures = {executor.submit(wrapper, uid): uid for uid in user_ids}
    for future in as_completed(futures):
        uid = futures[future]
        try:
            data = future.result()
            if data:
                results.append(data)
                print(f"用户{uid}获取成功")
        except Exception as e:
            print(f"用户{uid}异常：{e}")

3 错误重试与日志记录

网络不稳定时,需要重试机制，推荐使用 tenacity 库：

pip install tenacity

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 fetch_user_with_retry(user_id):
    return fetch_user(user_id)

日志记录建议用 logging 模块，而不是 print，方便后期排查：

import logging
logging.basicConfig(filename='api_batch.log', level=logging.INFO)
logging.info(f"用户{user_id}返回数据：{data}")

进阶技巧：异步调用与分布式任务

1 asyncio + aiohttp 方案（极致效率）

如果需要每秒处理数百次请求,建议用异步库 aiohttp：

import aiohttp
import asyncio
async def fetch(session, user_id):
    async with session.get(f"{API_BASE}/users/{user_id}", headers=HEADERS) as resp:
        return await resp.json()
async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, uid) for uid in range(1, 101)]
        results = await asyncio.gather(*tasks)
        print(len(results))
asyncio.run(main())

2 分布式任务队列（超大规模）

当数据量超过百万级别时,脚本已不适合，应使用 Celery + Redis 或 Prefect 等工作流引擎，把任务分发给多个机器并行处理。

常见问题与问答（FAQ）

Q1：API文档没有写明速率限制怎么办？ A：先尝试用二分法测试：设置每秒1次请求，成功后再逐渐增加频率，直到出现429状态码（Too Many Requests），实战中对免费接口最好留30%余量。

Q2：调用时总出现“连接超时”或“重置连接”？ A：可能是目标服务器对来源IP做了并发限制，解决方案：

增加重试次数（使用指数退避）
尝试使用代理池轮换IP
检查是否被WAF（防火墙）规则拦截

Q3：如何从CSV文件中读取参数进行批量调用？ A：使用 csv 模块或 pandas 库，例如用 pandas.read_csv('data.csv') 读取所有行，然后按行构造请求参数。

Q4：脚本运行中途断电了，如何断点续传？ A：写入文件时，每成功一条就记录到本地的success.json文件中，下次启动时，通过对比pending_ids和success_ids集合，只处理未完成的ID。（或用SQLite记录状态）

Q5：批量调用的结果如何保存？ A：小数据量用JSON文件；中等数据量用SQLite（自带Python支持）；大数据量建议直接写入MySQL或云存储（如Amazon S3）。

让批量调用成为生产力

批量调用API的核心在于三步走：

解耦逻辑：将请求参数、认证、速率限制作为独立配置
容错机制：重试+日志+进度记录
并发优化：根据API限制选择线程池或异步

建议新手先从单线程+简单sleep脚本做起，理解流程后再逐步优化，不必一开始就追求极致性能，稳定运行比快更重要，当你的脚本经过测试能在深夜自动处理完100万条数据，并准时发送报告到你的邮箱，你会感受到自动化带来的极大成就感。