Python批量API调用与限流策略：高效处理多源地理数据-Python教程-PHP中文网

Python批量API调用与限流策略：高效处理多源地理数据

本文详细介绍了如何使用Python处理来自多个列表的地理坐标数据，并通过API批量计算驾驶距离。核心内容包括利用zip函数高效迭代多组坐标，集成requests库进行API调用，以及通过自定义上下文管理器实现API请求的智能限流，确保程序稳定运行并遵守API服务条款。文章还强调了API响应错误处理的重要性，以构建健壮的数据获取流程。

1. 批量数据准备与API调用基础

在处理地理数据时，我们经常需要计算不同地点之间的距离。假设我们有四组列表数据，分别代表起始地点的纬度、经度以及目的地的纬度、经度，并希望通过一个外部api（例如osrm路由服务）来获取它们之间的驾驶距离。

首先，我们需要一个函数来封装对API的调用。这个函数接收两组坐标（起始点和目的点）并返回驾驶距离。

import requests
import json

def 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 requests
import json
from 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 contextmanager
from time import sleep
import requests
import json

# 全局变量，用于跟踪API调用次数
api_calls = 0

@contextmanager
def 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 DataFrame
df = 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需要密钥，切勿将密钥硬编码在代码中。应使用环境变量、配置文件或专门的密钥管理服务来安全地存储和访问密钥。