如何在循环中使用URL变量批量获取赛事API数据？

根据你的需求，以下是不同场景下的最优实现方案，涵盖同步、异步及关键优化策略：

1. 同步循环实现（小批量场景） #

适合race_keys数量较少（几十条以内）的情况，代码简单直观，无需额外依赖：

import requests

# 假设已通过前置API获取的race_keys列表
race_keys = ["race_20240520_001", "race_20240520_002", "race_20240520_003"]

def get_single_race_detail(race_key):
    """获取单场赛事详情"""
    api_url = f"https://your-api-domain.com/race/detail?race_key={race_key}"
    try:
        # 设置超时避免长时间阻塞
        resp = requests.get(api_url, timeout=10)
        # 主动触发HTTP错误（如4xx/5xx）
        resp.raise_for_status()
        return resp.json()
    except requests.exceptions.RequestException as e:
        print(f"赛事{race_key}详情采集失败: {str(e)}")
        return None

# 循环调用并收集有效数据
collected_details = []
for key in race_keys:
    detail = get_single_race_detail(key)
    if detail:
        collected_details.append(detail)

# 后续数据处理/存储逻辑
print(f"完成采集，共获取{len(collected_details)}条有效赛事详情")

2. 异步并发实现（大批量场景） #

当race_keys数量较多（上百条甚至更多）时，异步并发能大幅减少总耗时（API调用属于IO密集型任务，异步可充分利用网络等待时间），推荐使用aiohttp实现：

import aiohttp
import asyncio

race_keys = ["race_20240520_001", "race_20240520_002", "race_20240520_003"]

async def fetch_race_detail(session, race_key):
    """异步获取单场赛事详情"""
    api_url = f"https://your-api-domain.com/race/detail?race_key={race_key}"
    try:
        async with session.get(api_url, timeout=10) as resp:
            resp.raise_for_status()
            return await resp.json()
    except Exception as e:
        print(f"赛事{race_key}详情采集失败: {str(e)}")
        return None

async def main():
    # 创建客户端会话（复用连接，提升效率）
    async with aiohttp.ClientSession() as session:
        # 生成所有异步任务
        tasks = [fetch_race_detail(session, key) for key in race_keys]
        # 并发执行所有任务
        results = await asyncio.gather(*tasks)
        # 过滤失败的请求结果
        valid_details = [res for res in results if res is not None]
        
        print(f"完成采集，共获取{len(valid_details)}条有效赛事详情")
        # 此处可添加数据持久化逻辑（如写入数据库、JSON文件）

if __name__ == "__main__":
    asyncio.run(main())

3. 关键优化策略 #

错误重试机制 #

针对临时网络波动、API限流等问题，可通过重试机制提高采集成功率，推荐使用tenacity库：

from tenacity import retry, stop_after_attempt, wait_exponential

# 装饰器实现：最多重试3次，间隔时间指数递增（2s→4s→8s）
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
async def fetch_race_detail(session, race_key):
    # 原有异步逻辑不变
    api_url = f"https://your-api-domain.com/race/detail?race_key={race_key}"
    async with session.get(api_url, timeout=10) as resp:
        resp.raise_for_status()
        return await resp.json()

速率与并发控制 #

避免因并发过高触发API限流，可通过信号量限制同时发起的请求数：

async def main():
    # 限制同时最多5个并发请求
    semaphore = asyncio.Semaphore(5)
    
    async def limited_fetch(key):
        async with semaphore:
            return await fetch_race_detail(session, key)
    
    async with aiohttp.ClientSession() as session:
        tasks = [limited_fetch(key) for key in race_keys]
        results = await asyncio.gather(*tasks)
        # ...后续处理

实时数据持久化 #

采集过程中及时将数据写入文件或数据库，避免程序崩溃导致数据丢失：

import json

# 在获取valid_details后写入JSON文件
with open("daily_race_details.json", "w", encoding="utf-8") as f:
    json.dump(valid_details, f, ensure_ascii=False, indent=2)

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