📚 编程进阶教程站

第 20 章 · 综合实战：GitHub API 客户端 ​

本章目标：综合运用所学知识，构建一个完整的 GitHub API 客户端。

20.1 项目结构 ​

github_client/
├── client.py        # 主客户端
├── exceptions.py    # 自定义异常
├── config.py        # 配置管理
└── tests/
    └── test_client.py

20.2 异常定义 ​

python
# exceptions.py
class GitHubAPIError(Exception):
    """GitHub API 通用错误"""
    pass

class RateLimitError(GitHubAPIError):
    """速率限制错误"""
    def __init__(self, reset_time: int):
        self.reset_time = reset_time
        super().__init__(f'速率限制，{reset_time} 秒后重置')

class AuthenticationError(GitHubAPIError):
    """认证失败"""
    pass

20.3 客户端实现 ​

python
# client.py
import httpx
from typing import Optional, Dict, List
from .exceptions import GitHubAPIError, RateLimitError, AuthenticationError

class GitHubClient:
    def __init__(self, token: Optional[str] = None):
        self.base_url = 'https://api.github.com'
        self.client = httpx.Client(
            base_url=self.base_url,
            headers={'Accept': 'application/vnd.github.v3+json'},
            timeout=10.0,
            limits=httpx.Limits(max_connections=50)
        )
        if token:
            self.client.headers['Authorization'] = f'token {token}'
    
    def get_user(self, username: str) -> Dict:
        resp = self.client.get(f'/users/{username}')
        self._handle_response(resp)
        return resp.json()
    
    def list_repos(self, username: str, page: int = 1, per_page: int = 30) -> List[Dict]:
        params = {'page': page, 'per_page': per_page}
        resp = self.client.get(f'/users/{username}/repos', params=params)
        self._handle_response(resp)
        return resp.json()
    
    def search_repos(self, query: str, sort: str = 'stars') -> Dict:
        params = {'q': query, 'sort': sort, 'order': 'desc'}
        resp = self.client.get('/search/repositories', params=params)
        self._handle_response(resp)
        return resp.json()
    
    def _handle_response(self, resp: httpx.Response):
        if resp.status_code == 403 and 'rate limit' in resp.text.lower():
            reset = int(resp.headers.get('X-RateLimit-Reset', 0))
            raise RateLimitError(reset)
        elif resp.status_code == 401:
            raise AuthenticationError('认证失败')
        elif resp.status_code >= 400:
            raise GitHubAPIError(f'API 错误: {resp.status_code}')

20.4 测试 ​

python
# tests/test_client.py
import pytest
import respx
import httpx
from github_client import GitHubClient

@respx.mock
def test_get_user():
    respx.get("https://api.github.com/users/octocat").mock(
        return_value=httpx.Response(200, json={"login": "octocat"})
    )
    
    client = GitHubClient()
    user = client.get_user("octocat")
    assert user["login"] == "octocat"

@respx.mock
def test_rate_limit_error():
    respx.get("https://api.github.com/users/test").mock(
        return_value=httpx.Response(
            403,
            json={"message": "rate limit exceeded"},
            headers={"X-RateLimit-Reset": "1609459200"}
        )
    )
    
    client = GitHubClient()
    with pytest.raises(Exception) as exc_info:
        client.get_user("test")
    assert "rate limit" in str(exc_info.value).lower()

20.5 使用示例 ​

python
from github_client import GitHubClient

# 匿名访问（有限制）
client = GitHubClient()
user = client.get_user('torvalds')
print(f'{user["name"]} 有 {user["public_repos"]} 个公开仓库')

# 认证访问（更高限额）
client = GitHubClient(token='your-token')
repos = client.list_repos('github', per_page=10)
for repo in repos:
    print(f'{repo["name"]}: {repo["stargazers_count"]} stars')

20.6 本章小结 ​

  • 综合运用 Session、超时、异常处理、异步等知识；
  • 封装统一的错误处理层；
  • 编写完整的测试覆盖。

本章小结 ​

  • 综合运用 requests/httpx 构建生产级 API 客户端；
  • 异常层次化设计便于上层处理；
  • respx 提供可靠的测试 Mock。

🧪 随堂测验

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

1. GitHub API 未认证请求的速率限制是？

2. 处理 403 速率限制时应检查哪个响应头？

3. respx.mock 装饰器的作用是？

4. 生产级 API 客户端应该包含哪些特性？

🛠️ 动手实践 ​

  1. 为 GitHubClient 添加 get_issues() 方法，查询用户的所有 Issues。
  2. 添加缓存功能，对相同请求返回缓存结果。
  3. 编写完整的测试套件，覆盖正常和错误场景。