4. Python SDK 入门：使用 requests 库封装 API 调用，错误处理与重试机制

好，咱们进入实战环节。前面聊了那么多性能指标和监控理论，现在该动手了。这一章，我带你用 Python 的 requests 库，把 SDK 调用封装得既健壮又高效。

说白了，SDK 就是一套现成的工具包。你不需要自己拼 HTTP 请求、处理签名、解析响应。但现实是，很多 SDK 要么太笨重，要么文档不全。我个人习惯是，自己基于 requests 封装一层轻量级的调用层，灵活可控。

4.1 基础封装：从裸请求到函数化

先看一个最基础的版本。很多新手直接写 requests.get(url) 就完事了。这在 demo 里没问题，但上了生产环境，你会被坑得很惨。

import requests
import json

def call_api(url, method='GET', headers=None, data=None, timeout=10):
    """
    基础 API 调用封装
    :param url: 请求地址
    :param method: HTTP 方法
    :param headers: 请求头
    :param data: 请求体（字典）
    :param timeout: 超时时间（秒）
    :return: 响应 JSON 或 None
    """
    try:
        if method.upper() == 'GET':
            resp = requests.get(url, headers=headers, timeout=timeout)
        elif method.upper() == 'POST':
            resp = requests.post(url, headers=headers, json=data, timeout=timeout)
        else:
            raise ValueError(f"不支持的 HTTP 方法: {method}")

        resp.raise_for_status()  # 检查 HTTP 状态码
        return resp.json()

    except requests.exceptions.Timeout:
        print(f"[超时] 请求 {url} 超过 {timeout}s")
        return None
    except requests.exceptions.HTTPError as e:
        print(f"[HTTP错误] {e.response.status_code} - {url}")
        return None
    except requests.exceptions.ConnectionError:
        print(f"[连接错误] 无法连接到 {url}")
        return None
    except Exception as e:
        print(f"[未知错误] {str(e)}")
        return None

这段代码，我建议你直接复制到项目里当模板。它解决了三个基础问题：

• 超时控制：timeout=10 防止请求卡死
• 状态码检查：raise_for_status() 自动抛出 4xx/5xx 错误
• 异常分类：超时、HTTP 错误、连接错误分开处理

4.2 错误处理：别让一个报错拖垮整个系统

嗯，这里要注意。上面的基础封装有个大问题：它吞掉了所有错误。返回 None 会让调用方分不清是「接口返回空数据」还是「请求失败了」。

我在项目中遇到过这种情况：一个定时任务调了 10 个接口，其中一个超时返回了 None，结果后面的逻辑把 None 当正常数据处理，直接写入了数据库。排查了一下午才发现是超时问题。

更好的做法是：自定义异常 + 结构化错误信息。

class APIError(Exception):
    """API 调用自定义异常"""
    def __init__(self, message, status_code=None, response=None):
        super().__init__(message)
        self.status_code = status_code
        self.response = response

def call_api_safe(url, method='GET', headers=None, data=None, timeout=10):
    try:
        # ... 同上 ...
        resp.raise_for_status()
        return resp.json()

    except requests.exceptions.Timeout as e:
        raise APIError(f"请求超时: {url}", status_code=408) from e
    except requests.exceptions.HTTPError as e:
        status = e.response.status_code
        body = e.response.text[:200]  # 截取前200字符
        raise APIError(f"HTTP {status}: {body}", status_code=status, response=e.response) from e
    except requests.exceptions.ConnectionError as e:
        raise APIError(f"连接失败: {url}", status_code=503) from e

这样，调用方就能用 try-except 精确捕获错误，并拿到状态码和响应体。你想想看，排查问题时，HTTP 502: Bad Gateway 和 None 哪个更有用？

4.3 重试机制：别傻等，要有策略

网络请求总会失败。这是常态。关键是怎么重试。

我曾经见过一个同事，用 while True 加 time.sleep(1) 做重试。结果某个接口连续失败了 1000 次，日志刷了 10 万行，直接把磁盘打满了。嗯，这显然不行。

专业的重试策略，要考虑三点：

• 重试次数：最多重试几次？3 次？5 次？
• 退避策略：每次重试间隔多久？固定间隔？递增？
• 可重试条件：哪些错误值得重试？4xx 重试也没用，5xx 和超时才有意义。

我推荐用 tenacity 库，它把重试逻辑封装得特别优雅。当然，你也可以自己写，但没必要重复造轮子。

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),           # 最多重试3次
    wait=wait_exponential(multiplier=1, min=1, max=10),  # 指数退避：1s, 2s, 4s...
    retry=retry_if_exception_type((requests.exceptions.Timeout,
                                   requests.exceptions.ConnectionError,
                                   APIError)),  # 只重试特定异常
    reraise=True                            # 重试耗尽后抛出原异常
)
def call_api_with_retry(url, method='GET', headers=None, data=None, timeout=10):
    return call_api_safe(url, method, headers, data, timeout)

4.4 完整封装：一个生产级的 SDK 调用模块

把上面的东西整合起来，再加上日志和指标采集，就是一个能直接上线的模块了。

import logging
import time
import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

logger = logging.getLogger(__name__)

class APIClient:
    """轻量级 API 客户端"""

    def __init__(self, base_url, default_timeout=10, max_retries=3):
        self.base_url = base_url.rstrip('/')
        self.default_timeout = default_timeout
        self.max_retries = max_retries
        self.session = requests.Session()  # 复用连接，提升性能

    def _build_url(self, endpoint):
        return f"{self.base_url}/{endpoint.lstrip('/')}"

    def _log_request(self, method, url, duration, status):
        logger.info(f"[API] {method} {url} | {duration:.2f}s | {status}")

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
        retry=retry_if_exception_type((requests.exceptions.Timeout,
                                       requests.exceptions.ConnectionError)),
        reraise=True
    )
    def request(self, method, endpoint, **kwargs):
        url = self._build_url(endpoint)
        timeout = kwargs.pop('timeout', self.default_timeout)
        start = time.time()

        try:
            resp = self.session.request(method, url, timeout=timeout, **kwargs)
            resp.raise_for_status()
            duration = time.time() - start
            self._log_request(method, url, duration, resp.status_code)
            return resp.json()

        except requests.exceptions.HTTPError as e:
            duration = time.time() - start
            self._log_request(method, url, duration, e.response.status_code)
            # 4xx 不重试，直接抛出
            if 400 <= e.response.status_code < 500:
                raise
            # 5xx 让装饰器重试
            raise requests.exceptions.ConnectionError(f"Server error: {e.response.status_code}")

    def get(self, endpoint, params=None, **kwargs):
        return self.request('GET', endpoint, params=params, **kwargs)

    def post(self, endpoint, json=None, **kwargs):
        return self.request('POST', endpoint, json=json, **kwargs)

这个 APIClient 类，我用了 requests.Session()。为什么？因为 Session 会复用 TCP 连接，在频繁调用时能减少握手开销。我在压测时对比过，用 Session 比每次新建连接快了 30% 以上。

4.5 知识结构图：SDK 调用核心流程

下面这张图，概括了整个 SDK 调用的生命周期。你可以把它贴在工位旁边，写代码时对照着看。

4.6 实战建议：从封装到监控

最后，给你几个我在生产环境中用到的建议：

1. 加个熔断开关：如果某个接口连续失败超过 10 次，直接熔断 30 秒，不再发起请求。防止雪崩。
2. 记录每次调用的耗时：把 duration 上报到监控系统。一旦发现 P99 延迟飙升，立刻告警。
3. 不要全局捕获所有异常：在顶层（比如 Flask 路由或 Celery 任务）统一处理异常，底层只抛出自定义异常。
4. 测试重试逻辑：用 unittest.mock 模拟网络故障，验证重试次数和退避时间是否符合预期。