本文旨在指导如何使用Python高效地处理批量API请求，特别是当输入数据来源于多个列表时。我们将重点探讨如何将这些数据整合、如何通过自定义上下文管理器实现API请求的速率限制，以及如何确保请求的健壮性，通过错误处理机制提升代码的可靠性，最终将结果结构化为Pandas DataFrame。
1. 批量处理多列表数据与API交互
在许多数据处理场景中，我们需要从多个独立的列表中获取数据，并将这些数据作为参数传递给API进行批量查询。例如，计算一系列起点和终点之间的驾驶距离，其中起点和终点的经纬度分别存储在不同的列表中。
假设我们有四个列表：location_latitudes、location_longitudes、station_latitudes 和 station_longitudes。我们的目标是为每一对（起点经纬度，终点经纬度）调用API，并获取驾驶距离。
首先，定义一个基础的API调用函数。这里以Open Source Routing Machine (OSRM) API为例，它提供免费的路径规划服务。
import requestsimport jsonimport pandas as pdfrom contextlib import contextmanagerfrom time import sleepimport http# 定义全局变量用于API调用计数，以便后续限流使用api_calls_counter = 0def get_driving_distance(lat1, lon1, lat2, lon2): """ 通过OSRM API获取两点之间的驾驶距离。 参数: lat1 (float): 起点纬度 lon1 (float): 起点经度 lat2 (float): 终点纬度 lon2 (float): 终点经度 返回: float: 驾驶距离 (英里) """ # OSRM API的经纬度顺序是 经度,纬度 api_url = f"http://router.project-osrm.org/route/v1/car/{lon1},{lat1};{lon2},{lat2}?overview=false" try: response = requests.get(api_url) # 检查HTTP响应状态码，如果不是2xx，则抛出HTTPError response.raise_for_status() routes = json.loads(response.content) # 确保API返回了有效的路线数据 if routes and "routes" in routes and len(routes["routes"]) > 0: route_info = routes["routes"][0] driving_distance_meters = route_info.get('distance') if driving_distance_meters is not None: # 将米转换为英里 (1英里 = 1609.34米) return driving_distance_meters / 1609.34 else: print(f"警告: API响应中未找到距离信息 for {lat1},{lon1};{lat2},{lon2}") return None else: print(f"警告: API响应中未找到有效路线 for {lat1},{lon1};{lat2},{lon2}. 响应: {routes}") return None except requests.exceptions.HTTPError as e: print(f"HTTP错误: {e.response.status_code} - {e.response.text} for {lat1},{lon1};{lat2},{lon2}") return None except requests.exceptions.ConnectionError as e: print(f"连接错误: 无法连接到API服务 for {lat1},{lon1};{lat2},{lon2}. 错误: {e}") return None except requests.exceptions.Timeout as e: print(f"请求超时: API请求超时 for {lat1},{lon1};{lat2},{lon2}. 错误: {e}") return None except json.JSONDecodeError as e: print(f"JSON解析错误: 无法解析API响应 for {lat1},{lon1};{lat2},{lon2}. 错误: {e}") return None except Exception as e: print(f"发生未知错误: {e} for {lat1},{lon1};{lat2},{lon2}") return None
2. 实现API请求速率限制
在进行大量API调用时，为了避免触发API服务端的限流机制或被封禁IP，我们通常需要控制请求的频率。Python的contextlib.contextmanager提供了一种优雅的方式来实现这一点。
立即学习“Python免费学习笔记（深入）”；
以下是一个简单的速率限制上下文管理器，它会在达到指定调用次数后暂停一段时间：
@contextmanagerdef rate_limited(limit=500, delay=5): """ 一个上下文管理器，用于限制API调用速率。 在达到指定调用次数后，暂停一段时间。 参数: limit (int): 在暂停前允许的最大API调用次数。 delay (int): 暂停的秒数。 """ global api_calls_counter # 检查是否即将达到限制，如果是，则暂停并重置计数器 if api_calls_counter + 1 >= limit: print(f"达到API调用上限 {limit} 次，暂停 {delay} 秒...") sleep(delay) api_calls_counter = 0 # 重置计数器，或者可以 api_calls_counter -= limit # 增加API调用计数 api_calls_counter += 1 yield # 执行被包装的代码块
现在，我们将这个速率限制集成到 get_driving_distance 函数中：
# 更新后的 get_driving_distance 函数，包含速率限制def get_driving_distance_with_rate_limit(lat1, lon1, lat2, lon2): """ 通过OSRM API获取两点之间的驾驶距离，并集成速率限制。 """ with rate_limited(): # 使用速率限制上下文管理器 api_url = f"http://router.project-osrm.org/route/v1/car/{lon1},{lat1};{lon2},{lat2}?overview=false" try: response = requests.get(api_url) response.raise_for_status() # 检查HTTP状态码 routes = json.loads(response.content) if routes and "routes" in routes and len(routes["routes"]) > 0: route_info = routes["routes"][0] driving_distance_meters = route_info.get('distance') if driving_distance_meters is not None: return driving_distance_meters / 1609.34 else: print(f"警告: API响应中未找到距离信息 for {lat1},{lon1};{lat2},{lon2}") return None else: print(f"警告: API响应中未找到有效路线 for {lat1},{lon1};{lat2},{lon2}. 响应: {routes}") return None except requests.exceptions.HTTPError as e: print(f"HTTP错误: {e.response.status_code} - {e.response.text} for {lat1},{lon1};{lat2},{lon2}") return None except requests.exceptions.RequestException as e: # 捕获所有requests相关的异常 print(f"请求错误: {e} for {lat1},{lon1};{lat2},{lon2}") return None except json.JSONDecodeError as e: print(f"JSON解析错误: 无法解析API响应 for {lat1},{lon1};{lat2},{lon2}. 错误: {e}") return None except Exception as e: print(f"发生未知错误: {e} for {lat1},{lon1};{lat2},{lon2}") return None
3. 构建结果DataFrame
为了将原始输入数据和计算出的距离整合在一起，Pandas DataFrame是理想的选择。我们可以将四个输入列表和计算出的距离作为列，构建一个清晰的表格。
# 示例数据location_latitudes = [51.5074, 51.5100, 51.5200]location_longitudes = [-0.1278, -0.1300, -0.1400]station_latitudes = [51.5000, 51.5050, 51.5150]station_longitudes = [-0.1000, -0.1100, -0.1200]# 确保所有列表长度一致if not (len(location_latitudes) == len(location_longitudes) == len(station_latitudes) == len(station_longitudes)): raise ValueError("所有输入列表的长度必须一致。")# 存储计算结果driving_distances = []# 遍历所有数据对并调用APIfor i in range(len(location_latitudes)): lat1 = location_latitudes[i] lon1 = location_longitudes[i] lat2 = station_latitudes[i] lon2 = station_longitudes[i] distance = get_driving_distance_with_rate_limit(lat1, lon1, lat2, lon2) driving_distances.append(distance)# 创建DataFramedata = { 'Location_Latitude': location_latitudes, 'Location_Longitude': location_longitudes, 'Station_Latitude': station_latitudes, 'Station_Longitude': station_longitudes, 'Driving_Distance_Miles': driving_distances}df = pd.DataFrame(data)print("n最终结果DataFrame:")print(df)
4. 健壮的错误处理
在与外部API交互时，网络问题、API服务不稳定或返回异常数据是常见情况。因此，实现健壮的错误处理至关重要。
在 get_driving_distance_with_rate_limit 函数中，我们已经集成了以下错误处理机制：
response.raise_for_status(): 这是 requests 库提供的一个便捷方法。如果HTTP响应的状态码是4xx（客户端错误）或5xx（服务器错误），它会自动抛出 requests.exceptions.HTTPError 异常。这比手动检查 response.status_code 更简洁。try…except 块:requests.exceptions.RequestException: 这是所有 requests 库相关异常的基类（包括 HTTPError, ConnectionError, Timeout 等）。捕获此异常可以统一处理网络连接、超时等问题。json.JSONDecodeError: 当API返回的响应内容不是有效的JSON格式时，json.loads() 会抛出此异常。Exception: 捕获所有其他未预料到的错误，确保程序不会因小问题而崩溃。
在每个 except 块中，我们打印出详细的错误信息，并返回 None，以便在DataFrame中标记出未能成功获取距离的记录，便于后续分析和处理。
总结与注意事项
通过上述步骤，我们构建了一个完整的Python解决方案，用于批量调用API、处理多列表数据、实施速率限制并确保代码的健壮性。
数据整合: 使用循环遍历多列表数据，或考虑使用 zip 函数（如果列表数量固定且需要并行迭代）来优雅地处理输入。速率限制: contextlib.contextmanager 提供了一种非常Pythonic的方式来封装资源管理逻辑，例如API调用计数和暂停。错误处理: 始终对API请求进行错误处理。requests.Response.raise_for_status() 是一个很好的起点，结合 try-except 块可以处理各种网络和数据解析异常。数据存储: Pandas DataFrame是组织和展示批量API结果的理想工具，它易于后续分析和导出。API密钥: 如果使用的API需要认证（如API密钥），请务必将密钥安全地存储和管理，避免硬编码在代码中。异步请求: 对于非常大量的API调用，可以考虑使用 asyncio 和 aiohttp 等库实现异步请求，以进一步提高效率，但这会增加代码的复杂性。重试机制: 对于临时的网络问题或API服务波动，可以考虑实现一个简单的重试机制，即在失败后等待一小段时间再重新尝试请求。
通过遵循这些最佳实践，您可以构建出高效、稳定且易于维护的批量API调用应用程序。
以上就是Python批量API请求处理：数据整合、限流与错误管理的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1366982.html
微信扫一扫 
支付宝扫一扫 
