HTTP客户端封装与请求处理
大约 10 分钟
HTTP客户端封装与请求处理
前言：网络请求的"艺术"
还记得我刚开始做接口测试时，每次发送HTTP请求都要写一大堆重复代码，就像每次做饭都要从洗菜开始一样繁琐。后来我意识到，好的HTTP客户端封装就像一个贴心的厨师助手，把复杂的准备工作都做好了，让你专注于创造美味。
今天我们就来深入探讨这个pytest框架中的HTTP客户端实现，看看它是如何优雅地处理各种复杂的网络请求场景的。这不是简单的requests库使用教程，而是基于真实项目经验的深度封装实践。
BaseClient设计理念：简单而强大
设计目标
"""
BaseClient的设计目标
就像设计一辆好车，既要操作简单，又要性能强劲
"""
design_goals = {
"易用性": "简单的API，复杂的功能",
"可靠性": "自动重试，异常处理",
"可观测性": "详细日志，性能监控",
"可扩展性": "插件机制，自定义扩展",
"高性能": "连接池，并发支持"
}核心特性一览
# src/client/base_client.py 核心特性
class BaseClient:
"""
基础HTTP客户端 - 网络请求的瑞士军刀
特性清单：
✅ 自动重试机制
✅ 连接池管理
✅ 请求/响应日志
✅ 超时控制
✅ 会话管理
✅ 认证支持
✅ 异常处理
✅ 性能监控
"""
def __init__(self, base_url: str, timeout: Optional[int] = None):
self.base_url = base_url.rstrip('/')
self.timeout = timeout or config.get('API.timeout', 30)
self.session = requests.Session()
# 初始化各个组件
self._setup_default_headers()
self._setup_retry_strategy()
self._setup_connection_pool()
self._setup_hooks()
logger.info(f"🚀 HTTP客户端初始化完成: {self.base_url}")会话管理：保持连接的"记忆"
1. 连接池配置
def _setup_connection_pool(self):
"""
配置连接池 - 就像停车场管理
合理分配连接资源，提高复用效率
"""
from requests.adapters import HTTPAdapter
# 连接池配置
pool_config = {
'pool_connections': config.get('HTTP.pool_connections', 10),
'pool_maxsize': config.get('HTTP.pool_maxsize', 20),
'max_retries': 0, # 重试由我们自己控制
'pool_block': False
}
# 创建适配器
http_adapter = HTTPAdapter(**pool_config)
https_adapter = HTTPAdapter(**pool_config)
# 挂载适配器
self.session.mount('http://', http_adapter)
self.session.mount('https://', https_adapter)
logger.debug(f"连接池配置完成: {pool_config}")
def _setup_default_headers(self):
"""设置默认请求头 - 就像穿衣服的基本搭配"""
default_headers = {
'User-Agent': f'pytest-framework/{self._get_version()}',
'Accept': 'application/json',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive',
'Cache-Control': 'no-cache'
}
# 从配置文件读取自定义头
custom_headers = config.get('HTTP.default_headers', {})
default_headers.update(custom_headers)
self.session.headers.update(default_headers)
logger.debug(f"默认请求头设置完成: {list(default_headers.keys())}")
def _get_version(self) -> str:
"""获取框架版本"""
try:
from importlib.metadata import version
return version('pytest-framework')
except:
return '1.0.0'2. 智能重试机制
def _setup_retry_strategy(self):
"""
配置重试策略 - 就像坚持不懈的快递员
遇到问题不放弃，但也不会无限重试
"""
from requests.packages.urllib3.util.retry import Retry
retry_config = {
'total': config.get('HTTP.retry_times', 3),
'backoff_factor': config.get('HTTP.retry_delay', 1),
'status_forcelist': [429, 500, 502, 503, 504],
'method_whitelist': ['HEAD', 'GET', 'OPTIONS'],
'raise_on_status': False,
'raise_on_redirect': False
}
retry_strategy = Retry(**retry_config)
# 创建带重试的适配器
adapter = HTTPAdapter(max_retries=retry_strategy)
self.session.mount('http://', adapter)
self.session.mount('https://', adapter)
logger.debug(f"重试策略配置完成: {retry_config}")
def _should_retry(self, response: requests.Response, attempt: int) -> bool:
"""
判断是否应该重试
Args:
response: 响应对象
attempt: 当前尝试次数
Returns:
是否应该重试
"""
max_retries = config.get('HTTP.retry_times', 3)
if attempt >= max_retries:
return False
# 服务器错误或限流，可以重试
if response.status_code in [429, 500, 502, 503, 504]:
return True
# 网络超时，可以重试
if response.status_code == 408:
return True
return False认证处理：身份验证的"门卫"
1. 认证策略模式
# src/client/base_auth.py
"""
认证处理模块 - 身份验证的专家
支持多种认证方式，就像一个万能钥匙
"""
from abc import ABC, abstractmethod
from typing import Dict, Any
import base64
import hashlib
import hmac
import time
class AuthStrategy(ABC):
"""认证策略基类"""
@abstractmethod
def apply_auth(self, headers: Dict[str, str], **kwargs) -> Dict[str, str]:
"""应用认证信息到请求头"""
pass
class BearerTokenAuth(AuthStrategy):
"""Bearer Token认证"""
def __init__(self, token: str):
self.token = token
def apply_auth(self, headers: Dict[str, str], **kwargs) -> Dict[str, str]:
headers['Authorization'] = f'Bearer {self.token}'
return headers
class BasicAuth(AuthStrategy):
"""Basic认证"""
def __init__(self, username: str, password: str):
self.username = username
self.password = password
def apply_auth(self, headers: Dict[str, str], **kwargs) -> Dict[str, str]:
credentials = f'{self.username}:{self.password}'
encoded = base64.b64encode(credentials.encode()).decode()
headers['Authorization'] = f'Basic {encoded}'
return headers
class APIKeyAuth(AuthStrategy):
"""API Key认证"""
def __init__(self, api_key: str, header_name: str = 'X-API-Key'):
self.api_key = api_key
self.header_name = header_name
def apply_auth(self, headers: Dict[str, str], **kwargs) -> Dict[str, str]:
headers[self.header_name] = self.api_key
return headers
class SignatureAuth(AuthStrategy):
"""签名认证 - 适用于需要签名的API"""
def __init__(self, app_id: str, app_secret: str):
self.app_id = app_id
self.app_secret = app_secret
def apply_auth(self, headers: Dict[str, str], **kwargs) -> Dict[str, str]:
timestamp = str(int(time.time()))
method = kwargs.get('method', 'GET')
url = kwargs.get('url', '')
body = kwargs.get('body', '')
# 构建签名字符串
sign_string = f'{method}\n{url}\n{body}\n{timestamp}'
signature = hmac.new(
self.app_secret.encode(),
sign_string.encode(),
hashlib.sha256
).hexdigest()
headers.update({
'X-App-Id': self.app_id,
'X-Timestamp': timestamp,
'X-Signature': signature
})
return headers2. 认证管理器
class AuthManager:
"""认证管理器 - 统一管理各种认证方式"""
def __init__(self):
self._auth_strategy: Optional[AuthStrategy] = None
self._auto_refresh = False
self._token_expiry = None
def set_auth(self, auth_strategy: AuthStrategy, auto_refresh: bool = False):
"""设置认证策略"""
self._auth_strategy = auth_strategy
self._auto_refresh = auto_refresh
logger.info(f"认证策略已设置: {auth_strategy.__class__.__name__}")
def apply_auth(self, headers: Dict[str, str], **kwargs) -> Dict[str, str]:
"""应用认证信息"""
if self._auth_strategy is None:
return headers
# 检查token是否需要刷新
if self._auto_refresh and self._is_token_expired():
self._refresh_token()
return self._auth_strategy.apply_auth(headers, **kwargs)
def _is_token_expired(self) -> bool:
"""检查token是否过期"""
if self._token_expiry is None:
return False
return time.time() >= self._token_expiry
def _refresh_token(self):
"""刷新token"""
# 这里可以实现token刷新逻辑
logger.info("Token已刷新")
# 在BaseClient中集成认证
class BaseClient:
def __init__(self, base_url: str, timeout: Optional[int] = None):
# ... 其他初始化代码
self.auth_manager = AuthManager()
def set_auth(self, auth_strategy: AuthStrategy, auto_refresh: bool = False):
"""设置认证方式"""
self.auth_manager.set_auth(auth_strategy, auto_refresh)请求处理：精细化的流程控制
1. 请求预处理
def _prepare_request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
"""
请求预处理 - 就像做菜前的准备工作
把所有材料准备好，确保烹饪过程顺利
"""
# 构建完整URL
url = self._build_url(endpoint)
# 处理请求头
headers = kwargs.pop('headers', {})
headers = self.auth_manager.apply_auth(headers, method=method, url=url)
# 处理请求体
if 'json' in kwargs and 'data' in kwargs:
raise ValueError("不能同时指定json和data参数")
# 自动设置Content-Type
if 'json' in kwargs:
headers.setdefault('Content-Type', 'application/json')
elif 'data' in kwargs:
headers.setdefault('Content-Type', 'application/x-www-form-urlencoded')
# 处理超时
timeout = kwargs.pop('timeout', self.timeout)
# 处理代理
proxies = kwargs.pop('proxies', config.get('HTTP.proxies'))
# 处理SSL验证
verify = kwargs.pop('verify', config.get('HTTP.verify_ssl', True))
request_config = {
'method': method,
'url': url,
'headers': headers,
'timeout': timeout,
'proxies': proxies,
'verify': verify,
**kwargs
}
return request_config
def _build_url(self, endpoint: str) -> str:
"""
构建完整URL - 智能拼接
支持的格式：
- 相对路径: '/api/users'
- 绝对路径: 'https://api.example.com/users'
- 带参数的路径: '/api/users/{user_id}'
"""
if endpoint.startswith('http'):
return endpoint
# 处理路径参数
if '{' in endpoint and '}' in endpoint:
# 这里可以实现路径参数替换逻辑
pass
return f"{self.base_url}/{endpoint.lstrip('/')}"2. 响应处理与验证
def _process_response(self, response: requests.Response) -> requests.Response:
"""
响应处理 - 就像品尝师检查菜品质量
确保响应符合预期，提供有用的错误信息
"""
# 记录响应信息
self._log_response_details(response)
# 检查响应状态
if response.status_code >= 400:
self._handle_error_response(response)
# 验证响应格式
self._validate_response_format(response)
# 添加自定义属性
self._enhance_response(response)
return response
def _log_response_details(self, response: requests.Response):
"""记录详细的响应信息"""
duration = response.elapsed.total_seconds()
size = len(response.content)
logger.info(f"📥 响应详情: {response.status_code} | {duration:.3f}s | {size}bytes")
# 调试模式下记录更多信息
if config.get('DEBUG', False):
logger.debug(f"响应头: {dict(response.headers)}")
logger.debug(f"响应体: {response.text[:500]}...")
def _handle_error_response(self, response: requests.Response):
"""处理错误响应"""
error_info = {
'status_code': response.status_code,
'url': response.url,
'method': response.request.method,
'response_text': response.text[:1000]
}
# 根据状态码分类处理
if response.status_code == 401:
logger.error("🔐 认证失败，请检查认证信息")
elif response.status_code == 403:
logger.error("🚫 权限不足，请检查访问权限")
elif response.status_code == 404:
logger.error("🔍 资源不存在，请检查URL路径")
elif response.status_code >= 500:
logger.error("💥 服务器内部错误")
# 抛出自定义异常
from src.exceptions import HTTPError
raise HTTPError(response, error_info)
def _validate_response_format(self, response: requests.Response):
"""验证响应格式"""
content_type = response.headers.get('Content-Type', '')
# 验证JSON格式
if 'application/json' in content_type:
try:
response.json()
except ValueError as e:
logger.warning(f"⚠️ JSON格式无效: {e}")
# 验证字符编码
if response.encoding is None:
response.encoding = 'utf-8'
def _enhance_response(self, response: requests.Response):
"""增强响应对象 - 添加便利方法"""
def get_json_path(path: str, default=None):
"""使用JMESPath查询JSON数据"""
try:
import jmespath
data = response.json()
return jmespath.search(path, data) or default
except:
return default
def assert_status(self, expected_status: int):
"""断言状态码"""
if response.status_code != expected_status:
raise AssertionError(
f"期望状态码 {expected_status}，实际 {response.status_code}"
)
return response
def assert_json_path(self, path: str, expected_value):
"""断言JSON路径值"""
actual_value = get_json_path(path)
if actual_value != expected_value:
raise AssertionError(
f"路径 {path} 期望值 {expected_value}，实际值 {actual_value}"
)
return response
# 动态添加方法到响应对象
response.get_json_path = get_json_path
response.assert_status = lambda status: assert_status(response, status)
response.assert_json_path = lambda path, value: assert_json_path(response, path, value)高级特性：让客户端更智能
1. 请求钩子系统
def _setup_hooks(self):
"""设置请求钩子 - 在关键节点插入自定义逻辑"""
self._request_hooks = []
self._response_hooks = []
def add_request_hook(self, hook_func):
"""添加请求钩子"""
self._request_hooks.append(hook_func)
def add_response_hook(self, hook_func):
"""添加响应钩子"""
self._response_hooks.append(hook_func)
def _execute_request_hooks(self, request_config: Dict[str, Any]):
"""执行请求钩子"""
for hook in self._request_hooks:
try:
hook(request_config)
except Exception as e:
logger.warning(f"请求钩子执行失败: {e}")
def _execute_response_hooks(self, response: requests.Response):
"""执行响应钩子"""
for hook in self._response_hooks:
try:
hook(response)
except Exception as e:
logger.warning(f"响应钩子执行失败: {e}")
# 使用示例
def log_request_hook(request_config):
"""记录请求信息的钩子"""
logger.info(f"🚀 发送请求: {request_config['method']} {request_config['url']}")
def performance_hook(response):
"""性能监控钩子"""
duration = response.elapsed.total_seconds()
if duration > 3.0:
logger.warning(f"⚠️ 请求耗时过长: {duration:.3f}s")
# 注册钩子
client.add_request_hook(log_request_hook)
client.add_response_hook(performance_hook)2. 缓存机制
class ResponseCache:
"""响应缓存 - 避免重复请求"""
def __init__(self, max_size: int = 100, ttl: int = 300):
self.max_size = max_size
self.ttl = ttl
self._cache = {}
self._timestamps = {}
def _generate_key(self, method: str, url: str, params: Dict = None) -> str:
"""生成缓存键"""
import hashlib
key_data = f"{method}:{url}:{params or {}}"
return hashlib.md5(key_data.encode()).hexdigest()
def get(self, method: str, url: str, params: Dict = None) -> Optional[requests.Response]:
"""获取缓存的响应"""
key = self._generate_key(method, url, params)
if key not in self._cache:
return None
# 检查是否过期
if time.time() - self._timestamps[key] > self.ttl:
del self._cache[key]
del self._timestamps[key]
return None
logger.debug(f"💾 使用缓存响应: {method} {url}")
return self._cache[key]
def set(self, method: str, url: str, response: requests.Response, params: Dict = None):
"""缓存响应"""
if method.upper() != 'GET':
return # 只缓存GET请求
key = self._generate_key(method, url, params)
# 清理过期缓存
self._cleanup_expired()
# 限制缓存大小
if len(self._cache) >= self.max_size:
oldest_key = min(self._timestamps.keys(), key=lambda k: self._timestamps[k])
del self._cache[oldest_key]
del self._timestamps[oldest_key]
self._cache[key] = response
self._timestamps[key] = time.time()
logger.debug(f"💾 缓存响应: {method} {url}")
def _cleanup_expired(self):
"""清理过期缓存"""
current_time = time.time()
expired_keys = [
key for key, timestamp in self._timestamps.items()
if current_time - timestamp > self.ttl
]
for key in expired_keys:
del self._cache[key]
del self._timestamps[key]
# 在BaseClient中集成缓存
class BaseClient:
def __init__(self, base_url: str, timeout: Optional[int] = None):
# ... 其他初始化代码
self.cache = ResponseCache() if config.get('HTTP.enable_cache', False) else None
def request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
# 检查缓存
if self.cache and method.upper() == 'GET':
url = self._build_url(endpoint)
params = kwargs.get('params')
cached_response = self.cache.get(method, url, params)
if cached_response:
return cached_response
# 发送请求
response = self._send_request(method, endpoint, **kwargs)
# 缓存响应
if self.cache and method.upper() == 'GET' and response.status_code == 200:
url = self._build_url(endpoint)
params = kwargs.get('params')
self.cache.set(method, url, response, params)
return response实战应用示例
1. 业务API客户端
# tests/api_clients/user_client.py
"""
用户API客户端 - 业务层封装
基于BaseClient构建特定业务的API客户端
"""
from src.client.base_client import BaseClient
from src.client.base_auth import BearerTokenAuth
class UserAPIClient(BaseClient):
"""用户API客户端"""
def __init__(self, base_url: str, token: str = None):
super().__init__(base_url)
if token:
self.set_auth(BearerTokenAuth(token))
def create_user(self, user_data: Dict[str, Any]) -> Dict[str, Any]:
"""创建用户"""
response = self.post('/api/users', json=user_data)
response.assert_status(201)
return response.json()
def get_user(self, user_id: int) -> Dict[str, Any]:
"""获取用户信息"""
response = self.get(f'/api/users/{user_id}')
response.assert_status(200)
return response.json()
def update_user(self, user_id: int, user_data: Dict[str, Any]) -> Dict[str, Any]:
"""更新用户信息"""
response = self.put(f'/api/users/{user_id}', json=user_data)
response.assert_status(200)
return response.json()
def delete_user(self, user_id: int) -> bool:
"""删除用户"""
response = self.delete(f'/api/users/{user_id}')
response.assert_status(204)
return True
def search_users(self, keyword: str, page: int = 1, size: int = 10) -> Dict[str, Any]:
"""搜索用户"""
params = {'keyword': keyword, 'page': page, 'size': size}
response = self.get('/api/users/search', params=params)
response.assert_status(200)
return response.json()
# 使用示例
def test_user_operations():
"""用户操作测试"""
client = UserAPIClient('https://api.example.com', token='your-token')
# 创建用户
user_data = {'name': '张三', 'email': 'zhangsan@example.com'}
user = client.create_user(user_data)
# 获取用户
retrieved_user = client.get_user(user['id'])
assert retrieved_user['name'] == user_data['name']
# 更新用户
update_data = {'name': '李四'}
updated_user = client.update_user(user['id'], update_data)
assert updated_user['name'] == '李四'
# 删除用户
assert client.delete_user(user['id']) is True总结
HTTP客户端的封装就像打造一把趁手的工具，既要功能强大，又要使用简单。通过这篇文章，我们深入了解了：
- 设计理念：简单易用与功能强大的平衡
- 会话管理：连接池、重试机制、认证处理
- 请求处理：预处理、响应验证、错误处理
- 高级特性：钩子系统、缓存机制、性能优化
- 实战应用：业务API客户端的构建
这个HTTP客户端不仅仅是对requests库的简单封装，而是一个经过实战检验的、功能完整的网络请求解决方案。它让我们可以专注于业务逻辑的实现，而不用担心底层的网络细节。
下一篇文章，我们将探讨增强断言系统，看看如何使用JMESPath进行复杂的数据验证。