📚 课程目录第4章 API数据获取

4、API数据获取：调用RESTful API（认证、速率限制、分页），使用Python的requests库处理JSON/XML响应，案例：获取天气或股票数据

在数据采集的完整链路中，API（应用程序接口）数据获取是连接外部数据源与本地数据管道的关键环节。与直接爬取网页HTML不同，RESTful API提供了结构化、权限可控且通常带有明确速率限制的数据访问方式。本章将深入讲解如何通过Python的requests库高效、合规地调用RESTful API，重点覆盖认证机制、速率限制策略、分页处理以及JSON/XML响应解析，并通过天气与股票数据的实战案例，帮助学员建立从“拿到数据”到“稳定获取数据”的工程化思维。

4.1 RESTful API 核心原理与请求生命周期

REST（Representational State Transfer）是一种基于HTTP协议的架构风格，其核心思想是将后端数据资源抽象为URL端点，通过HTTP方法（GET、POST、PUT、DELETE）对资源进行操作。在数据采集场景中，我们最常使用的是GET请求，用于读取数据资源。一个完整的API请求生命周期包含以下阶段：

构造请求：确定基础URL（Base URL）、路径参数（Path Parameters）、查询参数（Query Parameters）和请求头（Headers）。
认证授权：在请求头中携带、Bearer 或通过OAuth流程获取临时凭证。
发送请求：通过requests.get()或session.get()发送HTTP请求，并设置超时时间。
接收响应：服务器返回HTTP状态码（200成功、401未认证、403禁止访问、429限流、500服务器错误）和响应体。
解析数据：根据Content-Type头（application/json或application/xml）解析响应体为Python数据结构。
错误处理：捕获网络异常、HTTP错误、JSON解析错误，并实施重试策略。

关键认知：API调用不是“一次请求拿全部数据”，而是“多次请求、分页拉取、限速控制”的工程过程。一个稳健的数据采集脚本，其代码量中约60%用于处理异常、重试、限速和日志记录，仅40%用于实际的数据解析。

4.2 认证机制：从到OAuth 2.0

绝大多数公开数据API要求调用者进行身份认证，以追踪使用量、实施配额控制并防止滥用。常见的认证方式有三种：

认证方式	传输位置	适用场景	安全性	Python实现要点
（查询参数）	URL查询字符串	低敏感度数据、快速接入	中低（Key可能被日志记录）	`params={'appid': key}`
（请求头）	HTTP Header	大多数商业API	中	`headers={'X-': key}`
Bearer / JWT	Authorization Header	OAuth 2.0流程、高安全要求	高（有时效性）	`headers={'Authorization': f'Bearer {}'}`

在实际开发中，推荐使用requests.Session()对象来管理认证信息和连接池。Session会自动保持请求头、Cookie，并复用底层TCP连接，显著提升多次请求的效率。以下是一个带认证的Session初始化示例：

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()
session.headers.update({
    'X-': 'your__here',
    'User-Agent': 'DataPipeline/1.0 (contact@example.com)'
})

# 配置重试策略：最多重试3次，对429和5xx状态码生效
retry_strategy = Retry(
    total=3,
    status_forcelist=[429, 500, 502, 503, 504],
    backoff_factor=1,  # 重试间隔：1s, 2s, 4s
    allowed_methods=["GET"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount('https://', adapter)
session.mount('http://', adapter)

安全警示：切勿将硬编码在脚本中或提交到Git仓库。推荐使用环境变量（os.environ['']）或配置文件（如.env文件）管理密钥，并在部署时使用密钥管理服务（如AWS Secrets Manager、Vault）。

4.3 速率限制与合规请求策略

速率限制（Rate Limiting）是API提供方保护后端服务、确保公平使用的重要手段。常见的限制方式包括：

QPS限制：每秒最多请求次数（如10次/秒）。
每日配额：每天最多请求次数（如1000次/天）。
并发限制：同时进行的请求数上限。

当请求超过限制时，服务器通常返回429 Too Many Requests状态码，并在响应头中携带Retry-After字段，指示客户端需要等待的秒数。一个负责任的采集程序应当：

主动限速：在客户端控制请求频率，使用time.sleep()或令牌桶算法。
读取响应头：解析X-RateLimit-Remaining、X-RateLimit-Reset等自定义头，动态调整请求速率。
处理429响应：捕获429状态码，读取Retry-After，等待后重试。

以下是一个带限速控制的请求封装函数：

import time
import logging

def rate_limited_get(session, url, params=None, min_interval=0.5):
    """带最小请求间隔的GET请求"""
    time.sleep(min_interval)  # 控制请求频率
    response = session.get(url, params=params, timeout=10)
    
    if response.status_code == 429:
        retry_after = int(response.headers.get('Retry-After', 5))
        logging.warning(f'触发限速，等待 {retry_after} 秒后重试')
        time.sleep(retry_after)
        return rate_limited_get(session, url, params, min_interval)
    
    response.raise_for_status()  # 其他4xx/5xx错误抛出异常
    return response

经验建议：在开发调试阶段，建议将min_interval设置为1秒以上，避免因调试频繁请求导致IP被临时封禁。生产环境中，可根据API文档的QPS限制设置合理的间隔，通常为限制值的80%左右，留出安全余量。

4.4 分页处理：遍历完整数据集

当API返回的数据量较大时（如历史股票数据、多城市天气记录），服务端会采用分页机制，避免单次响应过大。常见的分页方式有三种：

分页类型	参数示例	响应示例	适用场景
基于页码（Offset）	`?page=2&limit=50`	`{"page": 2, "total_pages": 10}`	数据总量相对稳定
基于游标（Cursor）	`?cursor=abc123`	`{"next_cursor": "xyz789"}`	数据频繁更新、实时性要求高
基于时间戳	`?start_time=...&end_time=...`	`{"has_more": true}`	时间序列数据（股票、日志）

游标分页是当前最推荐的方式，因为它不受数据新增或删除的影响，能保证遍历的一致性。以下是一个游标分页的完整实现：

def fetch_all_pages(session, base_url, params=None, max_pages=100):
    """遍历所有分页数据"""
    all_data = []
    cursor = None
    page_count = 0
    
    while True:
        # 构造当前页的请求参数
        current_params = dict(params or {})
        if cursor:
            current_params['cursor'] = cursor
        
        response = rate_limited_get(session, base_url, params=current_params)
        data = response.json()
        
        # 提取当前页数据
        page_items = data.get('items', [])
        all_data.extend(page_items)
        page_count += 1
        
        # 检查是否有下一页
        cursor = data.get('next_cursor')
        if not cursor or page_count >= max_pages:
            break
    
    logging.info(f'分页遍历完成，共获取 {len(all_data)} 条记录，{page_count} 页')
    return all_data

4.5 响应解析：JSON与XML处理

现代RESTful API绝大多数返回JSON格式，但仍有部分传统系统（如SOAP风格API）使用XML。Python的requests库内置了JSON解析器，而XML则需要借助xml.etree.ElementTree或lxml库。

JSON响应处理

response = session.get('https://api.openweathermap.org/data/2.5/weather', params={
    'q': 'Beijing',
    'appid': ,
    'units': 'metric'
})
data = response.json()  # 自动解析JSON

# 提取关键字段
temperature = data['main']['temp']
humidity = data['main']['humidity']
weather_desc = data['weather'][0]['description']
print(f'北京天气：{weather_desc}，温度 {temperature}°C，湿度 {humidity}%')

XML响应处理

import xml.etree.ElementTree as ET

response = session.get('https://example.com/api/stock?symbol=AAPL')
root = ET.fromstring(response.content)

# 使用XPath-like语法查找元素
price = root.find('.//price').text
volume = root.find('.//volume').text
print(f'AAPL 价格：{price}，成交量：{volume}')

健壮性要点：在解析响应前，务必检查response.status_code和Content-Type头。对于JSON解析，建议使用try-except捕获json.JSONDecodeError；对于XML，注意处理命名空间（Namespace）问题，否则find()可能无法匹配元素。

4.6 实战案例：构建天气与股票数据采集管道

下面通过一个综合案例，演示如何将上述知识整合为一个可复用的数据采集模块。该案例模拟从OpenWeatherMap获取多城市天气数据，并从Alpha Vantage获取股票历史行情，最终将数据标准化后存储为CSV文件。

案例背景与约束

数据源：OpenWeatherMap（免费版限60次/分钟）、Alpha Vantage（免费版限5次/分钟、每天500次）。
目标：采集5个城市的实时天气，以及3只股票近30天的日线数据。
约束：必须遵守限速要求，失败请求自动重试，数据落盘格式统一。

实现步骤

import pandas as pd
from datetime import datetime, timedelta

# 配置
CITIES = ['Beijing', 'Shanghai', 'Shenzhen', 'New York', 'London']
STOCKS = ['AAPL', 'MSFT', 'GOOGL']
WEATHER_URL = 'https://api.openweathermap.org/data/2.5/weather'
STOCK_URL = 'https://www.alphavantage.co/query'

def collect_weather_data():
    """采集多城市天气数据"""
    weather_records = []
    for city in CITIES:
        try:
            resp = rate_limited_get(session, WEATHER_URL, 
                                   params={'q': city, 'appid': WEATHER_KEY, 'units': 'metric'},
                                   min_interval=1.0)
            data = resp.json()
            weather_records.append({
                'city': city,
                'temp_c': data['main']['temp'],
                'humidity': data['main']['humidity'],
                'pressure': data['main']['pressure'],
                'description': data['weather'][0]['description'],
                'timestamp': datetime.utcnow().isoformat()
            })
        except Exception as e:
            logging.error(f'获取 {city} 天气失败: {e}')
    return pd.DataFrame(weather_records)

def collect_stock_data():
    """采集股票历史日线数据"""
    stock_records = []
    for symbol in STOCKS:
        try:
            resp = rate_limited_get(session, STOCK_URL,
                                   params={
                                       'function': 'TIME_SERIES_DAILY',
                                       'symbol': symbol,
                                       '': STOCK_KEY,
                                       'outputsize': 'compact'  # 最近100个交易日
                                   },
                                   min_interval=12.0)  # Alpha Vantage限速5次/分钟
            data = resp.json()
            time_series = data.get('Time Series (Daily)', {})
            for date_str, values in time_series.items():
                stock_records.append({
                    'symbol': symbol,
                    'date': date_str,
                    'open': float(values['1. open']),
                    'high': float(values['2. high']),
                    'low': float(values['3. low']),
                    'close': float(values['4. close']),
                    'volume': int(values['5. volume'])
                })
        except Exception as e:
            logging.error(f'获取 {symbol} 股票数据失败: {e}')
    return pd.DataFrame(stock_records)

# 执行采集
weather_df = collect_weather_data()
stock_df = collect_stock_data()

# 数据标准化与落盘
weather_df.to_csv('weather_data.csv', index=False)
stock_df.to_csv('stock_data.csv', index=False)

print(f'天气数据：{len(weather_df)} 条记录')
print(f'股票数据：{len(stock_df)} 条记录')

结果分析与复盘

运行上述脚本后，我们获得了5条天气记录和约300条股票记录（3只股票×100天）。在调试过程中发现两个关键问题：

限速触发：Alpha Vantage的免费版限制为5次/分钟，初次运行时未设置min_interval，导致频繁收到429错误。通过将间隔调整为12秒，问题解决。
数据时区：天气API返回的时间戳为UTC，而股票数据使用纽约时间。在后续分析中，需要统一转换为北京时间（UTC+8）或使用pandas.to_datetime指定时区。

工程化建议：在实际项目中，建议将、城市列表、股票列表等配置项外置到config.yaml或环境变量中，避免修改代码。同时，为采集任务添加日志记录（logging模块）和异常告警（如发送邮件或钉钉通知），确保数据管道的可观测性。

4.7 常见误区与总结复盘

在API数据采集的实践中，初学者常犯以下错误，需要特别注意规避：

误区	后果	正确做法
忽略状态码检查	将错误页面当作数据解析，导致程序崩溃	始终调用`response.raise_for_status()`或手动检查`status_code`
不设置超时	网络异常时程序永久挂起	所有请求必须设置`timeout`参数（如10秒）
一次性拉取全量数据	触发限速、内存溢出	使用分页+限速策略，分批获取
硬编码密钥	密钥泄露风险	使用环境变量或密钥管理服务
忽略响应头中的限速信息	无法动态调整请求频率	解析`X-RateLimit-*`头，实现自适应限速