用 Python 调公开 API:从请求到限速的实战要点

2026-05-06 15 预计阅读时间:1 分钟
来源:realpython.com AI 摘要 原文链接

免责声明:本文为 AI 摘要整理,建议结合原文阅读。摘要可能省略上下文、版本差异或边界条件,不作为官方说明。

预计阅读时间:8 分钟

公开数据 API 是获取天气、汇率、政府数据等信息的快捷通道。Python 的 requests 库让调用变得简单,但"能跑"和"能稳定跑"之间还有几道坎——状态码处理、认证方式、限速策略,每一环踩坑都会让脚本在半夜静默失败。

下面逐项拆解这些关键环节,并给出可直接复用的代码模板。

发起请求与解析响应

最基础的调用只需要两行:

import requests

resp = requests.get("https://api.open-meteo.com/v1/forecast",
                    params={"latitude": 39.9, "longitude": 116.4,
                            "current_weather": True})
data = resp.json()
print(data["current_weather"]["temperature"])

params 让 URL 参数自动编码,不用手动拼 ?latitude=39.9&...。返回的 JSON 用 .json() 直接转成 dict,比手动 json.loads(resp.text) 更省一步。

但这里隐含一个前提:服务器真的返回了 JSON。如果返回 HTML 错误页,.json() 会抛异常。所以下一步必须先看状态码。

状态码:别假装一切正常

HTTP 状态码是服务器给你的第一句实话。常见几类:

范围 含义 典型值
2xx 成功 200 OK
3xx 重定向 301 永久跳转
4xx 客户端错误 401 未认证、403 禁止、404 不存在
5xx 服务端错误 500 内部错误、503 不可用

健壮的做法是先检查状态码,再解析内容

resp = requests.get(url, params=params)

if resp.status_code != 200:
    # 4xx 一般是调用方的问题,5xx 可能需要重试
    print(f"请求失败: {resp.status_code}{resp.text[:200]}")
    # 根据业务决定是抛异常、返回空值还是重试
    raise RuntimeError(f"API 返回 {resp.status_code}")

data = resp.json()

requests 还提供了一个快捷断言:

resp.raise_for_status()  # 非 2xx 直接抛 HTTPError

适合脚本快速失败的场景。如果要做更细粒度的分支(比如 404 返回默认值、503 触发重试),就手动判断 status_code

认证:把钥匙带对

公开 API 也经常需要认证,常见三种方式:

1. API Key 作为查询参数

resp = requests.get("https://example.com/api/data",
                    params={"api_key": "YOUR_KEY", "q": "keyword"})

最简单,但 Key 会出现在 URL 里,日志中可见,安全性较低。

2. API Key 放在 Header

headers = {"X-API-Key": "YOUR_KEY"}
resp = requests.get("https://example.com/api/data", headers=headers)

Key 不再暴露在 URL 中,是更规范的做法。

3. OAuth / Bearer Token

headers = {"Authorization": f"Bearer {access_token}"}
resp = requests.get("https://example.com/api/data", headers=headers)

适用于需要用户授权的场景(比如调用 GitHub API 读取私有仓库)。获取 access_token 本身通常需要一次 OAuth 流程,这里不展开。

实际项目中,Key 和 Token 不要硬编码。用环境变量:

import os

API_KEY = os.environ["MY_API_KEY"]
headers = {"X-API-Key": API_KEY}

部署时通过 CI 密钥或 .env 文件注入,代码里不留明文。

限速:别把门敲坏了

很多公开 API 限制请求频率——每分钟 60 次、每天 1000 次,超出就返回 429 Too Many Requests。不做限速的脚本跑一会儿就会被封。

最简单的做法是固定间隔:

import time

for page in range(1, 11):
    resp = requests.get(f"https://api.example.com/items?page={page}",
                        headers=headers)
    resp.raise_for_status()
    items = resp.json()
    process(items)
    time.sleep(1)  # 每秒最多 1 次,远低于 60/min 的限制

更精细的做法是读取响应头里的限速信息。很多 API 会返回剩余次数:

remaining = int(resp.headers.get("X-RateLimit-Remaining", 60))
reset_at = int(resp.headers.get("X-RateLimit-Reset", 0))

if remaining <= 5:
    wait = max(reset_at - time.time(), 0)
    print(f"剩余额度低,等待 {wait:.0f} 秒")
    time.sleep(wait)

遇到 429 时主动退让,比被永久封禁划算得多:

if resp.status_code == 429:
    retry_after = int(resp.headers.get("Retry-After", 60))
    print(f"触发限速,{retry_after} 秒后重试")
    time.sleep(retry_after)
    # 重新发起同一请求

完整模板:带重试与限速的 API 客户端

把上面几项组合起来,得到一个可复用的骨架:

import os
import time
import requests

API_KEY = os.environ.get("MY_API_KEY", "")
BASE_URL = "https://api.example.com/v1"

def call_api(path, params=None, max_retries=3):
    headers = {"X-API-Key": API_KEY}
    url = f"{BASE_URL}/{path}"

    for attempt in range(max_retries):
        resp = requests.get(url, params=params, headers=headers)

        # 成功,直接返回
        if resp.status_code == 200:
            return resp.json()

        # 限速,等待后重试
        if resp.status_code == 429:
            wait = int(resp.headers.get("Retry-After", 60))
            print(f"限速,等待 {wait}s(第 {attempt+1} 次)")
            time.sleep(wait)
            continue

        # 服务端临时故障,短暂等待后重试
        if resp.status_code >= 500:
            print(f"服务端错误 {resp.status_code},5s 后重试")
            time.sleep(5)
            continue

        # 4xx 客户端错误,重试没用,直接抛
        resp.raise_for_status()

    raise RuntimeError(f"请求 {url} 失败,已重试 {max_retries} 次")

# 使用示例
data = call_api("weather", params={"city": "beijing"})
print(data)

运行前设置环境变量:

export MY_API_KEY="your_actual_key"
python api_client.py

选用建议与避坑清单

  • 选库:简单调用用 requests;需要异步并发用 httpx;爬取大量页面考虑 aiohttp + 异步限速。
  • 超时:永远显式设置 timeout=(5, 30)(连接超时 5 秒,读取超时 30 秒),避免请求无限挂起。
  • 日志:至少记录请求 URL、状态码和耗时,排查问题时不用猜。
  • 缓存:开发阶段用本地文件缓存响应,避免反复请求同一数据触发限速。
  • 分页:返回大量数据的 API 通常分页,注意循环读取直到 next_page 为空或总数已满。
  • 编码:少数 API 返回非 UTF-8 内容,resp.encoding 可能需要手动修正。

公开 API 看起来门槛很低,一个 requests.get 就能拿到数据。但生产级脚本的区别在于:状态码不忽略、认证不硬编码、限速不侥幸、重试有上限。把这四项做扎实,脚本才能从"本地跑一次"升级到"长期稳定运行"。


相关推荐