公开数据 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 就能拿到数据。但生产级脚本的区别在于:状态码不忽略、认证不硬编码、限速不侥幸、重试有上限。把这四项做扎实,脚本才能从"本地跑一次"升级到"长期稳定运行"。