1. 批量处理多列表数据与API交互
在许多数据处理场景中，我们需要从多个独立的列表中获取数据，并将这些数据作为参数传递给API进行批量查询。例如，计算一系列起点和终点之间的驾驶距离，其中起点和终点的经纬度分别存储在不同的列表中。
假设我们有四个列表：location_latitudes、location_longitudes、station_latitudes 和 station_longitudes。我们的目标是为每一对（起点经纬度，终点经纬度）调用API，并获取驾驶距离。
首先，定义一个基础的API调用函数。这里以Open Source Routing Machine (OSRM) API为例，它提供免费的路径规划服务。
import requests
import json
import pandas as pd
from contextlib import contextmanager
from time import sleep
import http
# 定义全局变量用于API调用计数，以便后续限流使用
api_calls_counter = 0
def 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免费学习笔记（深入）”；
以下是一个简单的速率限制上下文管理器，它会在达到指定调用次数后暂停一段时间：
@contextmanager
def 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 = []
# 遍历所有数据对并调用API
for 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)
# 创建DataFrame
data = {
'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调用应用程序。
