脚本怎样批量调用接口

本文目录导读：

1. Python 脚本（最常用）
2. Bash Shell 脚本
3. Node.js 脚本
4. 高级特性封装
5. 实用工具函数
6. 选择建议

Python 脚本（最常用）

使用 requests + 批量处理

import requests
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
# 方法1：顺序批量调用
def batch_call_sequential(api_list):
    results = []
    for api in api_list:
        try:
            response = requests.get(api['url'], params=api.get('params'))
            results.append({
                'api': api['url'],
                'status': response.status_code,
                'data': response.json()
            })
        except Exception as e:
            print(f"Error: {api['url']} - {str(e)}")
    return results
# 方法2：并发批量调用（推荐）
def batch_call_concurrent(api_list, max_workers=5):
    def call_single_api(api):
        try:
            response = requests.get(
                api['url'], 
                params=api.get('params'),
                timeout=10
            )
            return {
                'api': api['url'],
                'status': response.status_code,
                'data': response.json()
            }
        except Exception as e:
            return {'api': api['url'], 'error': str(e)}
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(call_single_api, api): api for api in api_list}
        for future in as_completed(futures):
            results.append(future.result())
    return results
# 使用示例
apis = [
    {'url': 'https://api.example.com/users', 'params': {'id': 1}},
    {'url': 'https://api.example.com/posts', 'params': {'page': 1}},
    {'url': 'https://api.example.com/comments', 'params': {'post_id': 1}}
]
results = batch_call_concurrent(apis)
print(json.dumps(results, indent=2))

Bash Shell 脚本

#!/bin/bash
# API 列表文件格式：URL,METHOD,BODY
apis_file="apis.txt"
# 批量调用函数
batch_call_apis() {
    while IFS=',' read -r url method body; do
        echo "Calling: $url"
        # HTTP GET
        if [ "$method" = "GET" ]; then
            curl -s -o /dev/null -w "Status: %{http_code}\n" "$url"
        # HTTP POST
        elif [ "$method" = "POST" ]; then
            curl -s -X POST -H "Content-Type: application/json" \
                 -d "$body" "$url" | jq .
        fi
        # 添加延迟
        sleep 1
    done < "$apis_file"
}
# 或者使用 xargs 并发
parallel_batch_call() {
    cat "$apis_file" | xargs -I {} -P 5 sh -c '
        url=$(echo {} | cut -d"," -f1)
        method=$(echo {} | cut -d"," -f2)
        echo "Calling: $url"
        curl -s "$url" | jq .
    '
}
# 执行
batch_call_apis

Node.js 脚本

const axios = require('axios');
const fs = require('fs').promises;
// 批量调用函数
const batchCallAPI = async (apis) => {
    const results = [];
    // 并发控制
    const concurrency = 3;
    for (let i = 0; i < apis.length; i += concurrency) {
        const batch = apis.slice(i, i + concurrency);
        const promises = batch.map(api => 
            axios.get(api.url, { params: api.params })
                .then(response => ({
                    url: api.url,
                    status: response.status,
                    data: response.data
                }))
                .catch(error => ({
                    url: api.url,
                    error: error.message
                }))
        );
        const batchResults = await Promise.all(promises);
        results.push(...batchResults);
    }
    return results;
};
// 使用示例
const apis = [
    { url: 'https://api.example.com/users', params: { id: 1 } },
    { url: 'https://api.example.com/posts', params: { page: 1 } }
];
batchCallAPI(apis).then(results => {
    console.log(JSON.stringify(results, null, 2));
});

高级特性封装

import asyncio
import aiohttp
from typing import List, Dict, Any
import time
class BatchAPICaller:
    def __init__(self, max_concurrent=5, retry_times=3):
        self.max_concurrent = max_concurrent
        self.retry_times = retry_times
        self.semaphore = asyncio.Semaphore(max_concurrent)
    async def call_single_api(self, session, api_info):
        async with self.semaphore:
            for attempt in range(self.retry_times):
                try:
                    method = api_info.get('method', 'GET').lower()
                    async with getattr(session, method)(
                        api_info['url'],
                        params=api_info.get('params'),
                        json=api_info.get('data'),
                        headers=api_info.get('headers'),
                        timeout=aiohttp.ClientTimeout(total=10)
                    ) as response:
                        return {
                            'url': api_info['url'],
                            'status': response.status,
                            'data': await response.json(),
                            'time': time.time()
                        }
                except Exception as e:
                    if attempt == self.retry_times - 1:
                        return {
                            'url': api_info['url'],
                            'error': str(e),
                            'attempts': attempt + 1
                        }
                    await asyncio.sleep(1 * (attempt + 1))
    async def batch_call(self, apis: List[Dict[str, Any]]):
        async with aiohttp.ClientSession() as session:
            tasks = [self.call_single_api(session, api) for api in apis]
            return await asyncio.gather(*tasks)
    def run(self, apis):
        return asyncio.run(self.batch_call(apis))
# 使用
caller = BatchAPICaller(max_concurrent=10, retry_times=2)
results = caller.run(apis)

实用工具函数

import pandas as pd
from typing import List, Dict
def load_apis_from_csv(file_path: str) -> List[Dict]:
    """从CSV文件加载API配置"""
    df = pd.read_csv(file_path)
    return df.to_dict('records')
def save_results(results: List[Dict], file_path: str):
    """保存结果到文件"""
    df = pd.DataFrame(results)
    df.to_csv(file_path, index=False)
def call_with_rate_limit(api_list, requests_per_second=10):
    """带速率限制的调用"""
    import time
    from collections import deque
    results = []
    timestamps = deque()
    for api in api_list:
        # 速率控制
        now = time.time()
        while timestamps and now - timestamps[0] > 1:
            timestamps.popleft()
        if len(timestamps) >= requests_per_second:
            sleep_time = 1 - (now - timestamps[0])
            if sleep_time > 0:
                time.sleep(sleep_time)
        # 调用API
        response = requests.get(api['url'])
        results.append({
            'url': api['url'],
            'status': response.status_code
        })
        timestamps.append(time.time())
    return results

选择建议

• 数据量小、顺序执行：使用 Bash 脚本
• 需要复杂逻辑处理：使用 Python
• 前端或Node.js环境：使用 JavaScript
• 需要高并发：使用 Python asyncio 或 Node.js
• 需要重试机制：使用封装好的类库

需要我详细解释某个方法或根据你的具体场景定制脚本吗？