Chapter 17 · API Client Encapsulation Patterns

Chapter goals: Learn how to encapsulate reusable API clients to improve code organization.

17.1 Basic Encapsulation Pattern

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()

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

17.2 httpx Async Encapsulation

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 Error Handling Encapsulation

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'Resource not found: {path}') from e
            elif resp.status_code >= 500:
                raise APIError(f'Server error: {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 Chapter Summary

• Encapsulating client classes centrally manages authentication, timeout, and other configurations;
• Distinguish synchronous (requests) and asynchronous (httpx) implementations;
• Unified exception handling improves code robustness.

🧪 随堂测验

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

1. When encapsulating API clients, authentication info should?

2. Which httpx class should async API clients use?

3. What is the main purpose of unified exception handling?

4. When encapsulating clients, base_url should?

🛠️ Hands-on Practice

1. Encapsulate a GitHub API client implementing user query and repository listing.
2. Add retry logic, automatically retry 5xx errors 3 times.
3. Write unit tests using respx Mock API responses.

After completing the exercises, proceed to Next chapter: Security Best Practices.