如何在Python中实现动态数量API调用的asyncio.gather异步调用？

核心实现思路 #

针对动态数量的API调用，直接利用asyncio.gather()配合列表推导式生成协程对象即可，无需手动逐个编写调用语句。

基础实现代码 #

import asyncio

# 模拟实际的API调用函数，替换为你的真实请求逻辑即可
async def call_api(key):
    await asyncio.sleep(1)  # 模拟网络延迟
    return f"API返回结果：{key}"

async def batch_fetch(keys):
    # 为每个key生成对应的协程对象
    coroutine_list = [call_api(key) for key in keys]
    # 批量执行所有协程，等待全部完成后收集结果
    results = await asyncio.gather(*coroutine_list)
    return results

# 测试执行
if __name__ == "__main__":
    dynamic_keys = ["user_001", "user_002", "user_003", "user_004"]
    final_results = asyncio.run(batch_fetch(dynamic_keys))
    print(final_results)

进阶优化：错误处理 #

如果希望某个API调用失败时不中断其他请求的执行，可给asyncio.gather()传入return_exceptions=True参数，异常会被包装进结果列表，后续可单独处理：

async def batch_fetch_with_error_handling(keys):
    coroutine_list = [call_api(key) for key in keys]
    # 即使部分调用失败，仍会返回所有结果（异常对象会混入结果列表）
    results = await asyncio.gather(*coroutine_list, return_exceptions=True)
    
    # 遍历结果，区分正常返回和异常
    for key, result in zip(keys, results):
        if isinstance(result, Exception):
            print(f"Key {key} 请求失败：{str(result)}")
        else:
            print(f"Key {key} 请求成功：{result}")
    return results

进阶优化：并发限流 #

如果目标API有调用速率限制，可通过asyncio.Semaphore控制同时执行的请求数量：

async def call_api_with_limit(key, semaphore):
    async with semaphore:
        # 原API调用逻辑
        await asyncio.sleep(1)
        return f"API返回结果：{key}"

async def batch_fetch_with_rate_limit(keys, max_concurrent=5):
    # 限制同时最多5个请求执行
    semaphore = asyncio.Semaphore(max_concurrent)
    coroutine_list = [call_api_with_limit(key, semaphore) for key in keys]
    results = await asyncio.gather(*coroutine_list)
    return results

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