Python调用API超时问题：如何修改批量请求代码避免超时？

针对批量调用API超时的问题，可以从以下几个方向修改代码：

1. 添加超时重试机制 #

单次超时可能是临时网络波动或服务器负载波动，添加重试逻辑可自动重试失败请求，避免直接放弃。可以用第三方库实现，也可手动编写：

手动实现重试逻辑 #

def get_val(element):
    url = 'https://www.xxxx/yyy/api/search'
    headers = {'Content-Type': 'application/json'}
    param = {'q': element, 'page' : 500}
    max_retries = 3
    retry_count = 0
    
    while retry_count < max_retries:
        try:
            response = requests.get(url, headers=headers, params=param, timeout=(3.05, 27))
            response.raise_for_status()  # 触发HTTP状态码异常
            docs = response.json()['response']['docs']
            
            for result in docs:
                file.write("%s\t%s\n" % (element, result['short_form']))
            return  # 请求成功则退出循环
        except Timeout:
            retry_count += 1
            print(f'元素 {element} 超时，正在重试 {retry_count}/{max_retries}')
        except Exception as e:
            print(f'元素 {element} 处理出错: {str(e)}')
            break  # 非超时异常直接终止重试
    print(f'元素 {element} 经过 {max_retries} 次重试仍失败')

使用tenacity库简化重试 #

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from requests.exceptions import Timeout

@retry(stop=stop_after_attempt(3), 
       wait=wait_exponential(multiplier=1, min=2, max=10), 
       retry=retry_if_exception_type(Timeout))
def get_val(element):
    url = 'https://www.xxxx/yyy/api/search'
    headers = {'Content-Type': 'application/json'}
    param = {'q': element, 'page' : 500}
    
    response = requests.get(url, headers=headers, params=param, timeout=(3.05, 27))
    response.raise_for_status()
    docs = response.json()['response']['docs']
    
    for result in docs:
        file.write("%s\t%s\n" % (element, result['short_form']))

2. 控制请求频率 #

连续高频请求容易触发服务器限流，导致超时或被封禁，可在每次请求后添加短暂延迟：

import time

# 循环调用时添加延迟
for i in lst:
    for element in i:
        get_val(element)
        time.sleep(0.5)  # 每次请求后暂停0.5秒，可根据实际情况调整时长

3. 复用HTTP连接 #

显式配置连接池，减少重复建立连接的开销，提升请求效率：

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

# 创建会话并配置连接池与重试策略
session = requests.Session()
retry_strategy = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount('https://', adapter)

# 修改get_val函数使用会话
def get_val(element, session):
    url = 'https://www.xxxx/yyy/api/search'
    headers = {'Content-Type': 'application/json'}
    param = {'q': element, 'page' : 500}
    
    try:
        response = session.get(url, headers=headers, params=param, timeout=(3.05, 27))
        response.raise_for_status()
        docs = response.json()['response']['docs']
        
        for result in docs:
            file.write("%s\t%s\n" % (element, result['short_form']))
    except Timeout:
        print(f'元素 {element} 请求超时')

# 循环调用时传入会话
for i in lst:
    for element in i:
        get_val(element, session)
        time.sleep(0.5)

4. 改用异步请求提升效率 #

同步串行请求效率低，使用aiohttp实现异步请求，同时控制并发数，既提升速度又避免压垮服务器：

import asyncio
import aiohttp
import aiofiles

async def get_val_async(element, session, semaphore, file):
    url = 'https://www.xxxx/yyy/api/search'
    headers = {'Content-Type': 'application/json'}
    param = {'q': element, 'page' : 500}
    max_retries = 3
    retry_count = 0
    
    async with semaphore:
        while retry_count < max_retries:
            try:
                async with session.get(url, headers=headers, params=param, timeout=aiohttp.ClientTimeout(total=27, connect=3.05)) as response:
                    response.raise_for_status()
                    docs = await response.json()
                    docs = docs['response']['docs']
                    
                    for result in docs:
                        await file.write(f"{element}\t{result['short_form']}\n")
                return
            except asyncio.TimeoutError:
                retry_count += 1
                print(f'元素 {element} 超时，正在重试 {retry_count}/{max_retries}')
            except Exception as e:
                print(f'元素 {element} 处理出错: {str(e)}')
                break
        print(f'元素 {element} 经过 {max_retries} 次重试仍失败')

async def main(lst, output_path):
    # 控制最大并发数，比如同时最多10个请求
    semaphore = asyncio.Semaphore(10)
    async with aiohttp.ClientSession() as session:
        async with aiofiles.open(output_path, 'a', encoding='utf-8') as file:
            tasks = []
            for sub_list in lst:
                for element in sub_list:
                    task = asyncio.create_task(get_val_async(element, session, semaphore, file))
                    tasks.append(task)
            await asyncio.gather(*tasks)

# 运行异步主函数
asyncio.run(main(lst, 'output.txt'))

5. 记录失败请求 #

建议记录处理失败的元素，后续可单独重试，避免数据丢失：

failed_elements = []

# 在get_val的异常处理中添加记录
except Timeout:
    failed_elements.append(element)
    print(f'元素 {element} 请求超时')

# 循环结束后保存失败列表
with open('failed_elements.txt', 'w', encoding='utf-8') as f:
    for elem in failed_elements:
        f.write(f"{elem}\n")

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