告别重复造轮子：打造Python requests的智能请求封装库

每次调用API时手动拼接Headers和Token的日子该结束了。如果你已经厌倦了在代码中反复复制粘贴相似的请求代码，或者羡慕Postman那种一键配置Headers的便捷体验，那么是时候升级你的开发工具链了。本文将带你从零构建一个高度封装的Python请求类，支持自动Token管理、多参数格式适配和智能异常处理，让HTTP请求变得像发送电子邮件一样简单。

1. 为什么需要封装requests库？

在日常开发中，我们经常需要与各种API打交道。无论是调用内部微服务还是接入第三方平台，HTTP请求都是最基础的交互方式。Python的requests库虽然简洁易用，但在企业级开发中直接使用原生方法会导致大量重复代码，尤其是处理以下场景时：

  • Token管理 ：每次请求都需要手动添加Authorization头
  • 参数格式化 ：根据API要求切换JSON、Form-data等格式
  • 异常处理 ：对超时、认证失败等场景缺乏统一处理
  • 日志记录 ：难以集中管理请求日志和监控
# 典型的不封装代码 - 每个请求都需要重复配置
headers = {
    'Authorization': 'Bearer xxxxx',
    'Content-Type': 'application/json'
}
response = requests.get(
    'https://api.example.com/data',
    headers=headers,
    params={'page': 1},
    timeout=10
)

Postman之所以受欢迎，正是因为它解决了这些问题。我们的目标是在Python代码中实现类似的开发体验，同时保持代码的灵活性和可维护性。

2. 设计智能请求类的核心架构

一个优秀的请求封装类应该像瑞士军刀一样多功能，又像专业工具一样精准。下面是我们设计的核心功能矩阵：

功能模块 实现要点 技术方案
认证管理 自动处理Token生命周期 可刷新的JWT令牌存储
请求构造 支持多种参数格式 动态Content-Type检测
异常处理 统一错误捕获和重试机制 装饰器+自定义异常类
性能监控 请求耗时统计和限流 时间戳记录+令牌桶算法
日志记录 完整的请求/响应追踪 结构化日志输出

2.1 基础请求类实现

我们从最基础的封装开始，创建一个支持GET/POST方法的类骨架：

import requests
from typing import Optional, Dict, Any

class SmartRequest:
    def __init__(self, base_url: str = '', default_timeout: int = 30):
        self.base_url = base_url
        self.default_timeout = default_timeout
        self.session = requests.Session()
        self._token = None
        
    @property
    def token(self):
        """获取当前认证令牌"""
        return self._token
    
    @token.setter 
    def token(self, value: str):
        """设置认证令牌"""
        self._token = value
        if value:
            self.session.headers.update({'Authorization': f'Bearer {value}'})
        else:
            self.session.headers.pop('Authorization', None)

这个基础版本已经解决了Token自动管理的问题。使用时只需设置一次Token，后续所有请求都会自动携带：

client = SmartRequest('https://api.example.com')
client.token = 'your_access_token'  # 设置一次，全局生效
response = client.get('/users')  # 自动携带Token

3. 多参数格式的智能适配

现代API通常支持多种参数格式，我们的封装需要能够自动处理这些差异。以下是常见的三种参数格式及其特点：

  1. JSON格式 ： Content-Type: application/json
    • 适合复杂嵌套数据结构
    • 需要明确指定内容类型
  2. Form-data ： Content-Type: multipart/form-data
    • 适合文件上传和混合数据
    • 边界值自动生成
  3. URL编码表单 ： Content-Type: application/x-www-form-urlencoded
    • 适合简单键值对
    • 需要URL编码

我们扩展SmartRequest类，增加智能参数检测功能：

class SmartRequest:
    # ... 延续之前的代码 ...
    
    def _detect_content_type(self, data: Any, files: Any) -> str:
        """自动检测合适的Content-Type"""
        if files is not None:
            return 'multipart/form-data'
        elif isinstance(data, (dict, list)):
            return 'application/json'
        return 'application/x-www-form-urlencoded'
    
    def request(self, method: str, endpoint: str, 
               data: Optional[Dict] = None,
               files: Optional[Dict] = None,
               **kwargs) -> requests.Response:
        """统一请求方法"""
        url = f"{self.base_url}{endpoint}"
        content_type = self._detect_content_type(data, files)
        
        headers = kwargs.pop('headers', {})
        headers['Content-Type'] = content_type
        
        if content_type == 'application/json':
            kwargs['json'] = data
        else:
            kwargs['data'] = data
            
        return self.session.request(
            method=method.upper(),
            url=url,
            files=files,
            timeout=self.default_timeout,
            headers=headers,
            **kwargs
        )
    
    def get(self, endpoint: str, params: Optional[Dict] = None, **kwargs):
        return self.request('GET', endpoint, params=params, **kwargs)
    
    def post(self, endpoint: str, data: Optional[Dict] = None, **kwargs):
        return self.request('POST', endpoint, data=data, **kwargs)

现在客户端代码可以完全不关心参数格式问题：

# JSON请求
client.post('/users', data={'name': 'Alice', 'age': 25})

# 表单请求
client.post('/login', data={'username': 'admin', 'password': '123'})

# 文件上传
with open('report.pdf', 'rb') as f:
    client.post('/upload', files={'file': f})

4. 异常处理与重试机制

健壮的API客户端需要妥善处理各种异常情况。我们设计一个分层的异常处理策略：

常见异常分类：

  • 网络层异常（连接超时、DNS解析失败）
  • 应用层异常（401未授权、403禁止访问）
  • 业务逻辑异常（自定义错误码）
  • 数据校验异常（响应格式不符）

实现带指数退避的重试机制：

from time import sleep
from functools import wraps

class APIError(Exception):
    """自定义API异常基类"""
    pass

def retry(max_attempts=3, initial_delay=1):
    """重试装饰器"""
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            attempts = 0
            delay = initial_delay
            last_exception = None
            
            while attempts < max_attempts:
                try:
                    return f(*args, **kwargs)
                except (requests.Timeout, requests.ConnectionError) as e:
                    last_exception = e
                    attempts += 1
                    if attempts < max_attempts:
                        sleep(delay)
                        delay *= 2  # 指数退避
            raise APIError(f"请求失败，重试{max_attempts}次后仍然失败") from last_exception
        return wrapper
    return decorator

class SmartRequest:
    # ... 延续之前的代码 ...
    
    @retry(max_attempts=3)
    def request(self, method: str, endpoint: str, **kwargs):
        try:
            response = self._raw_request(method, endpoint, **kwargs)
            response.raise_for_status()  # 检查HTTP状态码
            return response
        except requests.HTTPError as e:
            if e.response.status_code == 401:
                self._handle_unauthorized()
            raise APIError(f"API请求失败: {e}") from e
    
    def _raw_request(self, method: str, endpoint: str, **kwargs):
        """原始请求方法，不包含重试逻辑"""
        # ... 实际请求实现 ...

5. 高级功能扩展

5.1 请求日志与监控

在生产环境中，详细的请求日志对调试和监控至关重要：

import logging

class SmartRequest:
    def __init__(self, *args, **kwargs):
        self.logger = logging.getLogger('smart_request')
        # ... 其他初始化 ...
    
    def request(self, method: str, endpoint: str, **kwargs):
        start_time = time.time()
        request_id = str(uuid.uuid4())
        
        self.logger.info(
            "请求开始 | ID: %s | %s %s",
            request_id, method.upper(), endpoint
        )
        
        try:
            response = self._raw_request(method, endpoint, **kwargs)
            elapsed = time.time() - start_time
            
            self.logger.info(
                "请求成功 | ID: %s | 耗时: %.2fs | 状态码: %d",
                request_id, elapsed, response.status_code
            )
            return response
        except Exception as e:
            self.logger.error(
                "请求失败 | ID: %s | 错误: %s",
                request_id, str(e)
            )
            raise

5.2 自动Token刷新

对于使用短期Token的系统，我们可以实现自动刷新逻辑：

class SmartRequest:
    # ... 延续之前的代码 ...
    
    def _handle_unauthorized(self):
        """处理401未授权错误"""
        if self._refresh_token:
            try:
                new_token = self._refresh_access_token()
                self.token = new_token
                return
            except Exception as e:
                self.logger.warning("Token刷新失败: %s", str(e))
        raise APIError("认证失败，请重新登录")
    
    def _refresh_access_token(self) -> str:
        """使用refresh_token获取新的access_token"""
        response = self._raw_request(
            'POST',
            '/auth/refresh',
            json={'refresh_token': self._refresh_token}
        )
        return response.json()['access_token']

6. 完整实现与使用示例

将上述所有功能整合，我们得到最终的SmartRequest类：

# smart_request.py
import requests
import logging
import time
import uuid
from typing import Optional, Dict, Any

class APIError(Exception):
    pass

class SmartRequest:
    def __init__(self, base_url: str = '', default_timeout: int = 30):
        self.base_url = base_url.rstrip('/')
        self.default_timeout = default_timeout
        self.session = requests.Session()
        self._token = None
        self._refresh_token = None
        self.logger = logging.getLogger('smart_request')
        
    def set_tokens(self, access_token: str, refresh_token: str = None):
        self.token = access_token
        self._refresh_token = refresh_token
        
    @property
    def token(self):
        return self._token
    
    @token.setter
    def token(self, value: str):
        self._token = value
        if value:
            self.session.headers.update({'Authorization': f'Bearer {value}'})
        else:
            self.session.headers.pop('Authorization', None)
    
    @retry(max_attempts=3)
    def request(self, method: str, endpoint: str, 
               data: Optional[Dict] = None,
               files: Optional[Dict] = None,
               **kwargs) -> requests.Response:
        """统一请求入口"""
        request_id = str(uuid.uuid4())
        url = f"{self.base_url}{endpoint}"
        content_type = self._detect_content_type(data, files)
        
        headers = kwargs.pop('headers', {})
        headers['Content-Type'] = content_type
        
        # 记录请求开始
        self.logger.debug(
            "请求开始 | ID: %s | %s %s | 参数: %s",
            request_id, method.upper(), url, kwargs
        )
        start_time = time.time()
        
        try:
            # 实际请求执行
            response = self._raw_request(
                method, url,
                data=data,
                files=files,
                headers=headers,
                **kwargs
            )
            response.raise_for_status()
            
            # 记录成功日志
            elapsed = time.time() - start_time
            self.logger.info(
                "请求成功 | ID: %s | 耗时: %.2fs | 状态码: %d",
                request_id, elapsed, response.status_code
            )
            return response
            
        except requests.HTTPError as e:
            if e.response.status_code == 401:
                self._handle_unauthorized()
                # 重试一次
                return self.request(method, endpoint, data, files, **kwargs)
            raise APIError(f"API请求失败: {e}") from e
            
        except Exception as e:
            self.logger.error(
                "请求异常 | ID: %s | 错误: %s",
                request_id, str(e)
            )
            raise APIError(f"请求发生异常: {e}") from e
    
    # ... 其他方法保持不变 ...

使用示例：

# 初始化客户端
client = SmartRequest('https://api.example.com/v1')
client.set_tokens(
    access_token='your_access_token',
    refresh_token='your_refresh_token'
)

# 调用API
try:
    # 获取用户列表
    users = client.get('/users', params={'active': True}).json()
    
    # 创建新用户
    new_user = client.post('/users', data={
        'name': 'Bob',
        'email': 'bob@example.com'
    }).json()
    
except APIError as e:
    print(f"API调用失败: {e}")

7. 与Postman的功能对比

Postman作为API测试的黄金标准，我们的Python实现能达到怎样的相似度？以下是关键功能对比：

功能特性 Postman实现方式 SmartRequest实现方式 优势对比
Token管理 环境变量/全局变量 类属性自动携带 代码更简洁
参数格式 手动选择标签页 自动检测 减少配置错误
异常处理 通过Tests脚本验证 内置重试和统一异常 更适合生产环境
请求组织 集合和文件夹 Python模块化 版本控制友好
环境切换 环境变量切换 不同类实例 类型安全
自动化测试 Collection Runner pytest集成 更适合CI/CD

实际项目中，两者可以互补使用：Postman用于快速测试和调试，SmartRequest用于生产代码集成。特别是在以下场景中，代码封装更具优势：

  • 批量操作 ：需要循环调用同一API多次时
  • 条件逻辑 ：根据响应结果决定后续请求路径时
  • 性能关键 ：需要减少网络开销的高频调用场景
  • 安全要求 ：需要严格保护Token等敏感信息的场景
# 批量创建用户的示例 - 代码封装的优势显现
def onboard_users(users_data):
    results = []
    for user in users_data:
        try:
            resp = client.post('/users', data=user)
            results.append(resp.json())
        except APIError as e:
            logger.error(f"用户创建失败: {user['email']} - {str(e)}")
            continue
    return results
Logo
亚马逊云科技技术品牌专区

更多推荐

cover

咕泡科技接待湖南师范大学人工智能系访企团队，共启暑期实习育人新篇章

avatar 亚马逊云科技技术品牌专区
cover

Havenlon｜行业观察：执行控制系统不是伪需求，而是软件安全走到尽头后的新边界

avatar 亚马逊云科技技术品牌专区
cover

YOLOv8猫狗品种识别检测系统（项目源码+YOLO数据集+模型权重+UI界面+python+深度学习+环境配置+目标检测）

avatar 亚马逊云科技技术品牌专区