第 17 章 · API 客户端封装模式

本章目标：学习如何封装可复用的 API 客户端，提升代码组织性。

17.1 基础封装模式

import requests
from typing import Optional, Dict, Any

class GitHubClient:
    def __init__(self, token: Optional[str] = None):
        self.session = requests.Session()
        self.base_url = 'https://api.github.com'
        if token:
            self.session.headers['Authorization'] = f'token {token}'
    
    def get_user(self, username: str) -> Dict[str, Any]:
        resp = self.session.get(f'{self.base_url}/users/{username}')
        resp.raise_for_status()
        return resp.json()
    
    def list_repos(self, username: str, limit: int = 30) -> list:
        params = {'per_page': limit}
        resp = self.session.get(
            f'{self.base_url}/users/{username}/repos',
            params=params
        )
        resp.raise_for_status()
        return resp.json()

# 使用
client = GitHubClient(token='your-token')
user = client.get_user('octocat')
print(user['public_repos'])

17.2 httpx 异步封装

import httpx
from typing import Optional

class AsyncAPIClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.api_key = api_key
    
    async def get(self, path: str, **kwargs) -> dict:
        headers = {'X-API-Key': self.api_key, **kwargs.pop('headers', {})}
        async with httpx.AsyncClient() as client:
            resp = await client.get(f'{self.base_url}{path}', headers=headers, **kwargs)
            resp.raise_for_status()
            return resp.json()
    
    async def post(self, path: str, json: dict = None, **kwargs) -> dict:
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f'{self.base_url}{path}',
                json=json,
                headers={'Content-Type': 'application/json'},
                **kwargs
            )
            resp.raise_for_status()
            return resp.json()

17.3 错误处理封装

import requests
from requests.exceptions import HTTPError, RequestException

class APIError(Exception):
    pass

class APIClient:
    def __init__(self, base_url: str, timeout: float = 10.0):
        self.base_url = base_url
        self.timeout = timeout
        self.session = requests.Session()
    
    def _request(self, method: str, path: str, **kwargs):
        url = f'{self.base_url}{path}'
        kwargs.setdefault('timeout', self.timeout)
        try:
            resp = self.session.request(method, url, **kwargs)
            resp.raise_for_status()
            return resp.json() if resp.content else None
        except HTTPError as e:
            if resp.status_code == 404:
                raise APIError(f'资源不存在: {path}') from e
            elif resp.status_code >= 500:
                raise APIError(f'服务器错误: {resp.status_code}') from e
            raise
    
    def get(self, path: str, **kwargs):
        return self._request('GET', path, **kwargs)
    
    def post(self, path: str, **kwargs):
        return self._request('POST', path, **kwargs)

17.4 本章小结

• 封装客户端类集中管理认证、超时等配置；
• 区分同步（requests）和异步（httpx）实现；
• 统一异常处理提升代码健壮性。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 封装 API 客户端时，认证信息应该？

2. 异步 API 客户端应该使用哪个 httpx 类？

3. 统一异常处理的主要目的是？

4. 封装客户端时，base_url 应该？

🛠️ 动手实践

1. 封装一个 GitHub API 客户端，实现用户查询和仓库列表功能。
2. 添加重试逻辑，对 5xx 错误自动重试 3 次。
3. 编写单元测试，使用 respx Mock API 响应。

完成练习后，进入下一章：安全最佳实践。