利用Python asyncio 和 aiohttp 优化并发分页API请求

import asyncio
import aiohttp
import json # 假设响应是JSON格式

async def load_url(client: aiohttp.ClientSession, url: str, header: dict, page: int, parameters: dict) -> dict:
    """
    异步加载单个分页URL的数据。
    Args:
        client: aiohttp客户端会话。
        url: 基础URL。
        header: 请求头。
        page: 当前页码。
        parameters: 其他请求参数。
    Returns:
        解析后的JSON数据。
    Raises:
        Exception: 如果HTTP状态码不是200或响应内容解析失败。
    """
    current_parameters = parameters.copy() # 避免修改原始字典
    current_parameters["page"] = page

    try:
        # 设置超时时间，防止长时间阻塞
        async with client.get(url, headers=header, params=current_parameters, timeout=300) as resp:
            if resp.status != 200:
                # 记录详细的错误信息
                error_text = await resp.text()
                raise Exception(f"请求失败，状态码: {resp.status}, 响应: {error_text}")

            data = await resp.json()
            return data
    except asyncio.TimeoutError:
        raise Exception(f"请求页码 {page} 超时")
    except aiohttp.ClientError as e:
        raise Exception(f"请求页码 {page} 发生客户端错误: {e}")
    except Exception as e:
        raise Exception(f"处理页码 {page} 发生未知错误: {e}")

async def add_task(parameters: dict, url: str, header: dict, total_page_number: int) -> list:
    """
    创建并管理所有分页请求的异步任务。
    Args:
        parameters: 基础请求参数。
        url: 基础URL。
        header: 请求头。
        total_page_number: 总页数。
    Returns:
        包含所有分页数据结果的列表。
    """
    # 限制并发连接数，例如设置为50，避免对外部API造成过大压力
    connector = aiohttp.TCPConnector(force_close=True, limit=50) 
    async with aiohttp.ClientSession(connector=connector) as client:
        tasks = []
        for page in range(1, total_page_number + 1):
            # 将每个load_url协程包装成一个任务
            tasks.append(
                asyncio.ensure_future(
                    load_url(client, url, header, page, parameters)
                )
            )
        # 并发执行所有任务，并等待它们全部完成
        list_result = await asyncio.gather(*tasks, return_exceptions=True) # return_exceptions=True 允许收集异常而不是中断整个gather

    # 处理结果，过滤掉异常
    successful_results = []
    for res in list_result:
        if isinstance(res, Exception):
            print(f"警告: 某个请求发生异常: {res}")
            # 可以根据需要选择重新尝试、记录日志或跳过
        else:
            successful_results.append(res)

    return successful_results

def get_response(parameters: dict, url: str, header: dict, total_page_number: int) -> list:
    """
    同步入口点，用于启动异步任务并获取结果。
    Args:
        parameters: 基础请求参数。
        url: 基础URL。
        header: 请求头。
        total_page_number: 总页数。
    Returns:
        包含所有分页数据结果的列表。
    """
    return asyncio.run(add_task(parameters, url, header, total_page_number))

# 示例用法 (请替换为您的实际URL、参数和页数)
if __name__ == "__main__":
    # 假设的示例数据
    example_url = "https://api.example.com/data"
    example_headers = {"Authorization": "Bearer your_token"}
    example_params = {"param1": "value1"}
    example_total_pages = 20 # 假设有20页数据

    print("开始获取数据...")
    try:
        all_data = get_response(example_params, example_url, example_headers, example_total_pages)
        print(f"成功获取 {len(all_data)} 页数据。")
        # print(json.dumps(all_data[0], indent=2)) # 打印第一页数据示例
    except Exception as e:
        print(f"数据获取过程中发生错误: {e}")