🐍 第16课：API调用

—— 连接世界的桥梁

🏆 天气API+分页+认证

✅ Python验证通过

📌 本课目标

掌握 REST API 调用模式
掌握 API 认证方式
掌握分页数据处理
掌握 API 封装最佳实践
实战：天气 API 完整调用

1️⃣ REST API 基础

import requests

# RESTful 风格
# GET    /api/users      获取列表
# GET    /api/users/123   获取详情
# POST   /api/users      创建
# PUT    /api/users/123   全量更新
# PATCH  /api/users/123   部分更新
# DELETE /api/users/123   删除

# 获取列表
resp = requests.get("https://httpbin.org/get")
assert resp.status_code == 200

# 查看响应信息
print(resp.status_code)      # 200
print(resp.headers)          # 响应头
print(resp.json())           # 解析JSON
print(resp.elapsed)          # 请求耗时
print(resp.encoding)         # 编码

2️⃣ API 认证方式

import requests

# 1. API Key（查询参数）
resp = requests.get("https://api.example.com/data",
    params={"api_key": "your_key"})

# 2. Bearer Token（请求头）
headers = {"Authorization": "Bearer your_token"}
resp = requests.get("https://api.example.com/data", headers=headers)

# 3. Basic Auth
from requests.auth import HTTPBasicAuth
resp = requests.get("https://api.example.com/data",
    auth=HTTPBasicAuth("username", "password"))
# 或简写
resp = requests.get("https://api.example.com/data",
    auth=("username", "password"))

# 4. OAuth2（使用 requests-oauthlib）
# pip install requests-oauthlib
# from requests_oauthlib import OAuth2Session

3️⃣ 分页数据处理

import requests
from typing import Generator

# 方式1: 页码分页
def fetch_all_pages(base_url: str, per_page: int = 100) -> Generator[list, None, None]:
    """获取所有分页数据"""
    page = 1
    while True:
        resp = requests.get(base_url, params={
            "page": page,
            "per_page": per_page,
        }, timeout=10)
        resp.raise_for_status()
        data = resp.json()

        items = data.get("items", data.get("data", []))
        if not items:
            break

        yield items
        page += 1

# 方式2: 游标/offset分页
def fetch_by_cursor(base_url: str) -> Generator[list, None, None]:
    """基于游标的分页"""
    cursor = None
    while True:
        params = {}
        if cursor:
            params["cursor"] = cursor
        resp = requests.get(base_url, params=params, timeout=10)
        data = resp.json()
        yield data.get("items", [])
        cursor = data.get("next_cursor")
        if not cursor:
            break

# 使用：收集所有数据
# all_items = []
# for page_items in fetch_all_pages("https://api.example.com/users"):
#     all_items.extend(page_items)

4️⃣ API 封装最佳实践

import requests
from typing import Any, Optional
import time

class APIClient:
    """通用 API 客户端封装"""

    def __init__(self, base_url: str, token: str = "",
                 timeout: int = 10, max_retries: int = 3):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.max_retries = max_retries
        self.session = requests.Session()
        if token:
            self.session.headers["Authorization"] = f"Bearer {token}"
        self.session.headers["Content-Type"] = "application/json"

    def _request(self, method: str, path: str,
                 params: dict = None, data: dict = None) -> Any:
        url = f"{self.base_url}/{path.lstrip('/')}"
        for attempt in range(1, self.max_retries + 1):
            try:
                resp = self.session.request(
                    method, url, params=params, json=data,
                    timeout=self.timeout,
                )
                resp.raise_for_status()
                return resp.json()
            except requests.exceptions.RequestException as e:
                if attempt == self.max_retries:
                    raise
                wait = 2 ** attempt
                print(f"请求失败({attempt}/{self.max_retries}): {e}, {wait}s后重试")
                time.sleep(wait)

    def get(self, path: str, params: dict = None) -> Any:
        return self._request("GET", path, params=params)

    def post(self, path: str, data: dict = None) -> Any:
        return self._request("POST", path, data=data)

    def put(self, path: str, data: dict = None) -> Any:
        return self._request("PUT", path, data=data)

    def delete(self, path: str) -> Any:
        return self._request("DELETE", path)

# 使用
# api = APIClient("https://api.example.com", token="your_token")
# users = api.get("/users", params={"page": 1})
# result = api.post("/users", data={"name": "张三"})

5️⃣ 实战：天气 API

import requests

class WeatherAPI:
    """Open-Meteo 天气 API（免费，无需key）"""
    BASE_URL = "https://api.open-meteo.com/v1"

    def __init__(self):
        self.session = requests.Session()

    def get_current(self, latitude: float, longitude: float) -> dict:
        """获取当前天气"""
        params = {
            "latitude": latitude,
            "longitude": longitude,
            "current_weather": True,
            "timezone": "auto",
        }
        resp = self.session.get(
            f"{self.BASE_URL}/forecast",
            params=params, timeout=10,
        )
        resp.raise_for_status()
        return resp.json()["current_weather"]

    def get_forecast(self, latitude: float, longitude: float,
                     days: int = 7) -> dict:
        """获取天气预报"""
        params = {
            "latitude": latitude,
            "longitude": longitude,
            "daily": "temperature_2m_max,temperature_2m_min,precipitation_sum",
            "forecast_days": days,
            "timezone": "auto",
        }
        resp = self.session.get(
            f"{self.BASE_URL}/forecast",
            params=params, timeout=10,
        )
        resp.raise_for_status()
        return resp.json()["daily"]

# 使用
# weather = WeatherAPI()
# beijing = weather.get_current(39.9, 116.4)
# print(f"北京当前温度: {beijing['temperature']}°C")
# forecast = weather.get_forecast(39.9, 116.4)
# for date, hi, lo in zip(forecast['time'], forecast['temperature_2m_max'], forecast['temperature_2m_min']):
#     print(f"{date}: {lo}°C ~ {hi}°C")

6️⃣ 验证脚本

#!/usr/bin/env python3
"""第16课 API调用验证"""
import requests

def test_httpbin_get():
    """GET API测试"""
    resp = requests.get("https://httpbin.org/get",
                        params={"foo": "bar"}, timeout=10)
    assert resp.status_code == 200
    data = resp.json()
    assert data["args"]["foo"] == "bar"
    print("✅ GET API测试通过")

def test_httpbin_post():
    """POST API测试"""
    resp = requests.post("https://httpbin.org/post",
                         json={"name": "test"}, timeout=10)
    assert resp.status_code == 200
    data = resp.json()
    assert data["json"]["name"] == "test"
    print("✅ POST API测试通过")

def test_httpbin_auth():
    """认证测试"""
    resp = requests.get("https://httpbin.org/basic-auth/user/pass",
                        auth=("user", "pass"), timeout=10)
    assert resp.status_code == 200
    data = resp.json()
    assert data["authenticated"] == True
    print("✅ 认证测试通过")

def test_httpbin_status():
    """状态码测试"""
    resp = requests.get("https://httpbin.org/status/404", timeout=10)
    assert resp.status_code == 404
    print("✅ 状态码测试通过")

def test_api_client():
    """API客户端封装测试"""
    class SimpleClient:
        def __init__(self):
            self.session = requests.Session()
        def get(self, path, **kw):
            resp = self.session.get(f"https://httpbin.org{path}", timeout=10, **kw)
            resp.raise_for_status()
            return resp.json()

    client = SimpleClient()
    data = client.get("/get", params={"test": "1"})
    assert data["args"]["test"] == "1"
    print("✅ API客户端封装测试通过")

if __name__ == "__main__":
    test_httpbin_get()
    test_httpbin_post()
    test_httpbin_auth()
    test_httpbin_status()
    test_api_client()
    print("\n🎉 第16课全部验证通过！")

✅ GET API测试通过 ✅ POST API测试通过 ✅ 认证测试通过 ✅ 状态码测试通过 ✅ API客户端封装测试通过 🎉 第16课全部验证通过！

🔑 本课要点

封装 API Client——统一处理认证、重试、错误
生成器处理分页——yield 逐页返回，内存友好
指数退避——2^n 秒重试间隔，避免雪崩
raise_for_status()——HTTP 错误立即抛异常
免费 API 练手——httpbin.org、open-meteo.com

← 上一课 📚 目录下一课 →