本文详细介绍了如何使用Python处理来自多个列表的地理坐标数据，并通过API批量计算驾驶距离。核心内容包括利用zip函数高效迭代多组坐标，集成requests库进行API调用，以及通过自定义上下文管理器实现API请求的智能限流，确保程序稳定运行并遵守API服务条款。文章还强调了API响应错误处理的重要性，以构建健壮的数据获取流程。
1. 批量数据准备与API调用基础
在处理地理数据时，我们经常需要计算不同地点之间的距离。假设我们有四组列表数据，分别代表起始地点的纬度、经度以及目的地的纬度、经度，并希望通过一个外部api（例如osrm路由服务）来获取它们之间的驾驶距离。
首先，我们需要一个函数来封装对API的调用。这个函数接收两组坐标（起始点和目的点）并返回驾驶距离。
import requestsimport jsondef get_driving_distance(lat1, lon1, lat2, lon2): """ 通过OSRM API计算两点之间的驾驶距离。 参数: lat1 (float): 起始点纬度 lon1 (float): 起始点经度 lat2 (float): 目的点纬度 lon2 (float): 目的点经度 返回: float: 驾驶距离（英里） """ # OSRM API的URL格式是 /lon1,lat1;lon2,lat2 url = f"http://router.project-osrm.org/route/v1/car/{lon1},{lat1};{lon2},{lat2}?overview=false" r = requests.get(url) # 解析API响应 routes = json.loads(r.content) # 检查API响应是否包含有效的路线信息 if "routes" in routes and len(routes["routes"]) > 0: route_info = routes["routes"][0] driving_distance_meters = route_info['distance'] # 将米转换为英里 (1 英里 ≈ 1609.34 米) return driving_distance_meters / 1609.34 else: print(f"API响应未包含有效路线: {routes}") return None # 或者抛出异常
当拥有多组坐标列表时，我们可以使用Python内置的zip函数将这些列表的对应元素打包成元组，然后进行迭代调用上述函数。
# 示例数据location_latitudes = [51.5074, 51.5100]location_longitudes = [-0.1278, -0.1300]station_latitudes = [51.5200, 51.5300]station_longitudes = [-0.1000, -0.1100]# 迭代并计算距离distances = []for lat1, lon1, lat2, lon2 in zip(location_latitudes, location_longitudes, station_latitudes, station_longitudes): distance = get_driving_distance(lat1, lon1, lat2, lon2) distances.append(distance)print(f"计算出的距离: {distances}")
2. API请求的健壮性与错误处理
在实际的API调用中，网络问题、API服务中断或请求参数错误都可能导致失败。为了使代码更加健壮，我们应该检查API响应的状态码。requests库提供了Response.raise_for_status()方法，它会在收到错误状态码（如4xx或5xx）时抛出HTTPError异常。或者，我们也可以直接检查Response.status_code属性。
import requestsimport jsonfrom http import HTTPStatus # 导入HTTPStatus枚举，提高可读性def get_driving_distance_robust(lat1, lon1, lat2, lon2): """ 通过OSRM API计算两点之间的驾驶距离，包含错误处理。 """ url = f"http://router.project-osrm.org/route/v1/car/{lon1},{lat1};{lon2},{lat2}?overview=false" try: r = requests.get(url) r.raise_for_status() # 如果状态码不是200，则抛出HTTPError routes = json.loads(r.content) if "routes" in routes and len(routes["routes"]) > 0: route_info = routes["routes"][0] driving_distance_meters = route_info['distance'] return driving_distance_meters / 1609.34 else: print(f"API响应未包含有效路线或数据: {routes}") return None except requests.exceptions.HTTPError as e: print(f"HTTP错误发生: {e.response.status_code} - {e.response.text}") return None except requests.exceptions.ConnectionError as e: print(f"网络连接错误: {e}") return None except json.JSONDecodeError as e: print(f"JSON解析错误: {e.msg} - 响应内容: {r.text if 'r' in locals() else 'N/A'}") return None except Exception as e: print(f"发生未知错误: {e}") return None
3. 实现API请求限流机制
频繁地调用API可能会触及服务提供商的请求限制，导致IP被封禁或服务暂时不可用。为了避免这种情况，我们可以实现一个简单的限流机制。Python的contextlib.contextmanager提供了一种优雅的方式来管理资源，非常适合实现这种“进入-执行-退出”的逻辑。
立即学习“Python免费学习笔记（深入）”；
我们将创建一个上下文管理器，它会跟踪API调用次数，并在达到预设限制时暂停执行一段时间。
from contextlib import contextmanagerfrom time import sleepimport requestsimport json# 全局变量，用于跟踪API调用次数api_calls = 0@contextmanagerdef rate_limited(limit=500, delay=5): """ 一个上下文管理器，用于实现API请求限流。 当API调用次数达到limit时，暂停delay秒。 参数: limit (int): 允许在不暂停的情况下进行的API调用次数上限。 delay (int): 达到上限后暂停的秒数。 """ global api_calls # 声明使用全局变量 # 在进入上下文之前检查是否需要暂停 if api_calls + 1 >= limit: print(f"达到 {limit} 次API调用限制，暂停 {delay} 秒...") sleep(delay) api_calls = 0 # 暂停后重置计数器 api_calls += 1 # 每次进入上下文时增加计数 yield # 执行被包装的代码块# 将限流机制集成到API调用函数中def get_driving_distance_rate_limited(lat1, lon1, lat2, lon2): """ 通过OSRM API计算两点之间的驾驶距离，集成限流和错误处理。 """ with rate_limited(limit=500, delay=5): # 使用限流上下文管理器 url = f"http://router.project-osrm.org/route/v1/car/{lon1},{lat1};{lon2},{lat2}?overview=false" try: r = requests.get(url) r.raise_for_status() routes = json.loads(r.content) if "routes" in routes and len(routes["routes"]) > 0: route_info = routes["routes"][0] driving_distance_meters = route_info['distance'] return driving_distance_meters / 1609.34 else: print(f"API响应未包含有效路线或数据: {routes}") return None except requests.exceptions.HTTPError as e: print(f"HTTP错误发生: {e.response.status_code} - {e.response.text}") return None except requests.exceptions.ConnectionError as e: print(f"网络连接错误: {e}") return None except json.JSONDecodeError as e: print(f"JSON解析错误: {e.msg} - 响应内容: {r.text if 'r' in locals() else 'N/A'}") return None except Exception as e: print(f"发生未知错误: {e}") return None
4. 整合与数据处理
现在，我们可以将所有组件整合起来，处理多组数据，并最终将结果整理成易于分析的格式，例如Pandas DataFrame。
import pandas as pd# 假设 get_driving_distance_rate_limited 函数已定义# 更多示例数据location_latitudes = [51.5074, 51.5100, 51.5000, 51.5050, 51.5150]location_longitudes = [-0.1278, -0.1300, -0.1100, -0.1200, -0.1400]station_latitudes = [51.5200, 51.5300, 51.5100, 51.5250, 51.5350]station_longitudes = [-0.1000, -0.1100, -0.0900, -0.1050, -0.1250]# 准备一个列表来存储所有结果results = []# 批量调用API并收集结果for i, (lat1, lon1, lat2, lon2) in enumerate(zip(location_latitudes, location_longitudes, station_latitudes, station_longitudes)): print(f"处理第 {i+1} 组数据: ({lat1}, {lon1}) -> ({lat2}, {lon2})") distance = get_driving_distance_rate_limited(lat1, lon1, lat2, lon2) # 将原始数据和计算出的距离存储起来 results.append({ 'Location_Lat': lat1, 'Location_Lon': lon1, 'Station_Lat': lat2, 'Station_Lon': lon2, 'Driving_Distance_Miles': distance })# 将结果转换为Pandas DataFramedf = pd.DataFrame(results)print("n最终结果DataFrame:")print(df)# 可以将DataFrame保存到CSV文件# df.to_csv("driving_distances.csv", index=False)
5. 注意事项与最佳实践
API服务条款： 在使用任何第三方API之前，务必仔细阅读其服务条款和使用限制。OSRM是一个开源路由引擎，通常用于自部署或低量请求。对于商业应用或高并发需求，可能需要考虑使用Google Maps API、Mapbox等提供商的付费服务，它们通常有更严格的限额和计费模式。异常处理的粒度： 针对不同类型的API错误（例如，404 Not Found、429 Too Many Requests、5xx Server Error），可以实现更细致的错误处理逻辑，例如重试机制、错误日志记录或通知。数据持久化： 如果需要处理大量数据，建议在处理过程中定期保存中间结果。这样即使程序中断，也无需从头开始，可以从上次保存的地方恢复。并发/异步请求： 对于极其庞大的数据集，单线程顺序调用API效率可能较低。可以考虑使用concurrent.futures库（例如ThreadPoolExecutor）或asyncio库来实现并发或异步请求，进一步提高数据获取速度，但需要注意并发请求对API限流策略的影响，可能需要更复杂的限流逻辑。API密钥管理： 如果使用的API需要密钥，切勿将密钥硬编码在代码中。应使用环境变量、配置文件或专门的密钥管理服务来安全地存储和访问密钥。
总结
通过本文的学习，我们掌握了如何利用Python高效地处理来自多个列表的数据，并通过API进行批量操作。核心技术包括：
使用zip函数优雅地迭代多组相关数据。利用requests库进行HTTP请求，并结合raise_for_status()进行健壮的错误处理。通过contextlib.contextmanager创建自定义上下文管理器，实现智能的API请求限流机制，以避免触发API限制并确保程序稳定运行。将处理结果整合到Pandas DataFrame中，便于后续的数据分析和存储。
这些技术不仅适用于地理数据处理，也适用于任何需要批量调用API的场景，是构建高效、可靠数据处理流程的关键。
以上就是Python批量API调用与限流策略：高效处理多源地理数据的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1366972.html
微信扫一扫 
支付宝扫一扫 
