当前位置：首页 > 文章列表 > 文章 > python教程 > Python高效API请求处理技巧

Python高效API请求处理技巧

2025-08-24 08:18:33 0浏览收藏

本文详细介绍了如何利用Python高效处理批量API请求，尤其是在处理来自多个列表的数据时。文章重点讲解了数据整合技巧，利用`contextlib.contextmanager`实现API请求的速率限制，有效避免触发API服务端的限流机制。同时，强调了通过完善的错误处理机制来提升代码的健壮性，确保程序在面对网络问题或API异常时依然能够稳定运行。最终，文章还指导如何将API请求的结果结构化为Pandas DataFrame，方便后续的数据分析与处理。无论您是数据分析师还是Python开发者，本文提供的技巧都能帮助您更好地处理批量API调用任务。

Python批量API请求处理：数据整合、限流与错误管理

本文旨在指导如何使用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 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提供了一种优雅的方式来实现这一点。

以下是一个简单的速率限制上下文管理器，它会在达到指定调用次数后暂停一段时间：

@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调用应用程序。

今天关于《Python高效API请求处理技巧》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！