企业官网建设流程全解析

1. 为什么需要封装requests请求？

在接口自动化测试项目中，直接使用原生requests库发送HTTP请求会面临几个典型问题。首先是代码重复度高，每个测试用例中都需要重复编写headers处理、异常捕获、日志记录等基础代码。其次是维护成本大，当接口鉴权方式变更或需要统一添加某个请求头时，需要在所有用例中逐个修改。

我曾参与过一个电商平台的接口测试项目，初期直接在用例中写requests.get()，后来当平台要求所有接口必须携带traceId时，团队不得不花费两天时间修改200多个测试文件。这种教训让我们意识到请求封装的重要性。

封装后的请求工具应当具备以下核心能力：

• 统一处理鉴权信息（如自动添加Authorization头）
• 自动记录请求日志和响应时间
• 内置重试机制应对网络抖动
• 支持常见的参数类型转换
• 提供简洁的调用方式

2. 基础请求封装实现

2.1 类结构设计

我们创建一个HttpClient类作为基础封装，采用单例模式确保全局唯一实例：

import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry class HttpClient: _instance = None def __new__(cls): if not cls._instance: cls._instance = super().__new__(cls) cls._instance._init_session() return cls._instance def _init_session(self): self.session = requests.Session() retry = Retry( total=3, backoff_factor=0.3, status_forcelist=[500, 502, 504] ) adapter = HTTPAdapter(max_retries=retry) self.session.mount('http://', adapter) self.session.mount('https://', adapter)

关键点：使用requests.Session()可以保持TCP连接复用，相比每次新建连接能提升30%以上的请求速度。Retry配置实现了自动重试机制。

2.2 GET/POST方法封装

基础请求方法需要处理以下细节：

def get(self, url, params=None, **kwargs): """ :param url: 请求地址 :param params: query参数(dict) :return: Response对象 """ start_time = time.time() try: response = self.session.get( url, params=params, **kwargs ) response.raise_for_status() return response except requests.exceptions.RequestException as e: self._log_error('GET', url, e, time.time()-start_time) raise

POST方法需要额外处理不同参数类型：

def post(self, url, data=None, json=None, **kwargs): """ :param data: form表单数据(dict) :param json: JSON格式数据(dict) """ if json is not None: kwargs['json'] = json headers = kwargs.get('headers', {}) headers.setdefault('Content-Type', 'application/json') kwargs['headers'] = headers elif data is not None: kwargs['data'] = data # 后续处理与GET类似...

3. 高级功能扩展

3.1 自动化鉴权处理

实际项目中常见的鉴权方式处理：

def __init__(self): self.token = None self.token_expire = 0 def _check_auth(self): if time.time() > self.token_expire - 300: # 提前5分钟刷新 self._refresh_token() def _refresh_token(self): auth_url = "https://api.example.com/auth" resp = self.session.post(auth_url, json={ "app_key": "your_key", "app_secret": "your_secret" }) self.token = resp.json()['token'] self.token_expire = time.time() + 3600 # 假设token有效期1小时

3.2 文件上传处理

针对multipart/form-data类型的封装：

def upload(self, url, files, data=None): """ :param files: {'file': ('filename', fileobj, 'content_type')} """ if not isinstance(files, dict): raise ValueError("files must be a dict") file_list = [] for name, file_info in files.items(): if len(file_info) == 2: file_list.append((name, (file_info[0], file_info[1]))) else: file_list.append((name, tuple(file_info))) return self.post(url, files=file_list, data=data)

4. 实战中的经验技巧

4.1 长整型参数处理

当需要传递long类型参数时，需要注意JSON序列化问题：

def post_json(url, data): # 自定义JSON编码器处理long类型 class CustomEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, (int, long)): return str(obj) return super().default(obj) return self.post(url, json=data, encoder=CustomEncoder)

4.2 429 Too Many Requests处理

应对接口限流的策略：

def _init_session(self): # 在原有retry配置基础上增加 retry = Retry( total=5, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE", "POST"] ) # ...其余初始化代码

4.3 请求日志优化

完整的日志记录应该包含：

def _log_request(self, method, url, params, data, headers): log_data = { 'method': method, 'url': url, 'time': datetime.now().isoformat() } if params: log_data['params'] = params if data: if isinstance(data, dict): log_data['data'] = {k: '******' if 'password' in k else v for k,v in data.items()} else: log_data['data'] = '<binary data>' logger.info(json.dumps(log_data, ensure_ascii=False))

5. 完整封装示例

结合所有功能的完整实现：

class HttpClient: def __init__(self, base_url=None): self.base_url = base_url.rstrip('/') if base_url else None self.session = requests.Session() self._setup_retry() self._setup_timeout() self.auth_handler = None def _setup_retry(self): retry = Retry( total=3, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=frozenset(['GET', 'POST']) ) adapter = HTTPAdapter(max_retries=retry) self.session.mount('http://', adapter) self.session.mount('https://', adapter) def request(self, method, path, **kwargs): url = f"{self.base_url}/{path.lstrip('/')}" if self.base_url else path if self.auth_handler: kwargs = self.auth_handler.add_auth(kwargs) self._log_request(method, url, kwargs) start = time.time() try: resp = self.session.request(method, url, **kwargs) resp.raise_for_status() return resp except Exception as e: self._log_error(method, url, e, time.time()-start) raise

使用示例：

client = HttpClient('https://api.example.com/v1') response = client.get('users', params={'page': 1}) data = client.post('orders', json={'product_id': 123})

在实际项目中，这样的封装可以使接口测试代码量减少40%以上，同时显著提升稳定性和可维护性。特别是在微服务架构下，当需要统一调整所有接口的超时时间或重试策略时，只需修改封装类的一处配置即可。