requests库深度应用与封装

前言：requests，HTTP请求的"人生苦短"

如果说Python的哲学是"人生苦短，我用Python"，那么requests库就完美诠释了这一理念。还记得我刚开始做接口测试时，用的是urllib库，那酸爽的体验至今难忘：

# urllib的痛苦回忆
import urllib.request
import urllib.parse
import json

data = urllib.parse.urlencode({'key': 'value'}).encode()
req = urllib.request.Request('http://example.com', data=data)
req.add_header('Content-Type', 'application/x-www-form-urlencoded')
response = urllib.request.urlopen(req)
result = json.loads(response.read().decode())

而用requests只需要：

import requests
response = requests.post('http://example.com', json={'key': 'value'})
result = response.json()

这就是requests的魅力：让HTTP请求变得优雅而简单。今天我们就来深入探讨requests的各种高级用法和封装技巧。

requests基础回顾

基本HTTP方法

import requests

# GET请求
response = requests.get('https://httpbin.org/get')

# POST请求
response = requests.post('https://httpbin.org/post', json={'key': 'value'})

# PUT请求
response = requests.put('https://httpbin.org/put', data={'key': 'value'})

# DELETE请求
response = requests.delete('https://httpbin.org/delete')

# PATCH请求
response = requests.patch('https://httpbin.org/patch', json={'key': 'value'})

请求参数处理

# URL参数
params = {'page': 1, 'size': 10, 'keyword': '测试'}
response = requests.get('https://api.example.com/search', params=params)
# 实际请求URL: https://api.example.com/search?page=1&size=10&keyword=%E6%B5%8B%E8%AF%95

# 请求头
headers = {
    'User-Agent': 'MyTestClient/1.0',
    'Authorization': 'Bearer your-token-here',
    'Content-Type': 'application/json'
}
response = requests.get('https://api.example.com/data', headers=headers)

# 请求体数据
# JSON数据
response = requests.post('https://api.example.com/users', json={'name': '张三', 'age': 25})

# 表单数据
response = requests.post('https://api.example.com/login', data={'username': 'admin', 'password': '123456'})

# 文件上传
files = {'file': open('test.txt', 'rb')}
response = requests.post('https://api.example.com/upload', files=files)

requests高级特性

1. Session对象：保持连接状态

Session对象就像一个有记忆的浏览器，可以保持cookies、连接池等状态：

import requests

# 创建Session
session = requests.Session()

# 设置默认请求头
session.headers.update({
    'User-Agent': 'MyTestClient/1.0',
    'Accept': 'application/json'
})

# 登录获取token
login_response = session.post('https://api.example.com/login', json={
    'username': 'admin',
    'password': '123456'
})

if login_response.status_code == 200:
    token = login_response.json()['token']
    # 设置认证头，后续请求都会携带
    session.headers.update({'Authorization': f'Bearer {token}'})

# 后续请求会自动携带认证信息
user_response = session.get('https://api.example.com/user/profile')
orders_response = session.get('https://api.example.com/user/orders')

# 记得关闭session
session.close()

2. 请求和响应钩子

钩子函数可以在请求发送前或响应接收后执行自定义逻辑：

import requests
import time

def log_request(request, *args, **kwargs):
    """请求日志钩子"""
    print(f"发送请求: {request.method} {request.url}")
    print(f"请求头: {request.headers}")

def log_response(response, *args, **kwargs):
    """响应日志钩子"""
    print(f"收到响应: {response.status_code}")
    print(f"响应时间: {response.elapsed.total_seconds()}s")

# 创建session并添加钩子
session = requests.Session()
session.hooks['response'].append(log_response)

# 发送请求时会自动调用钩子
response = session.get('https://httpbin.org/get')

3. 超时和重试机制

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

def create_session_with_retries():
    """创建带重试机制的session"""
    session = requests.Session()
    
    # 配置重试策略
    retry_strategy = Retry(
        total=3,                    # 总重试次数
        backoff_factor=1,           # 重试间隔倍数
        status_forcelist=[429, 500, 502, 503, 504],  # 需要重试的状态码
        method_whitelist=["HEAD", "GET", "OPTIONS"]   # 允许重试的方法
    )
    
    # 创建适配器
    adapter = HTTPAdapter(max_retries=retry_strategy)
    
    # 挂载适配器
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    
    return session

# 使用带重试的session
session = create_session_with_retries()

try:
    # 设置超时时间：连接超时5秒，读取超时10秒
    response = session.get('https://httpbin.org/delay/2', timeout=(5, 10))
    print(response.json())
except requests.exceptions.Timeout:
    print("请求超时")
except requests.exceptions.RequestException as e:
    print(f"请求异常: {e}")

4. 代理和SSL配置

import requests

# 代理配置
proxies = {
    'http': 'http://proxy.example.com:8080',
    'https': 'https://proxy.example.com:8080'
}

# SSL配置
response = requests.get(
    'https://api.example.com/data',
    proxies=proxies,
    verify=False,  # 忽略SSL证书验证（仅测试环境）
    cert=('client.cert', 'client.key')  # 客户端证书
)

封装实战：构建企业级HTTP客户端

基于多年的实践经验，我总结出了一套企业级HTTP客户端的封装方案：

1. 基础客户端类

import requests
import json
import time
from typing import Dict, Any, Optional
from urllib.parse import urljoin

class BaseHTTPClient:
    """基础HTTP客户端"""
    
    def __init__(self, base_url: str, timeout: int = 30):
        self.base_url = base_url.rstrip('/')
        self.timeout = timeout
        self.session = requests.Session()
        
        # 设置默认请求头
        self.session.headers.update({
            'User-Agent': 'AutoTestClient/1.0',
            'Accept': 'application/json',
            'Content-Type': 'application/json'
        })
        
        # 配置重试机制
        self._setup_retry_strategy()
    
    def _setup_retry_strategy(self):
        """配置重试策略"""
        from requests.adapters import HTTPAdapter
        from requests.packages.urllib3.util.retry import Retry
        
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504]
        )
        
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)
    
    def _build_url(self, endpoint: str) -> str:
        """构建完整URL"""
        return urljoin(self.base_url + '/', endpoint.lstrip('/'))
    
    def _log_request(self, method: str, url: str, **kwargs):
        """记录请求日志"""
        print(f"[REQUEST] {method.upper()} {url}")
        if 'json' in kwargs:
            print(f"[REQUEST BODY] {json.dumps(kwargs['json'], ensure_ascii=False, indent=2)}")
    
    def _log_response(self, response: requests.Response, start_time: float):
        """记录响应日志"""
        duration = time.time() - start_time
        print(f"[RESPONSE] {response.status_code} - {duration:.3f}s")
        
        try:
            response_data = response.json()
            print(f"[RESPONSE BODY] {json.dumps(response_data, ensure_ascii=False, indent=2)}")
        except:
            print(f"[RESPONSE BODY] {response.text[:500]}...")
    
    def request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
        """通用请求方法"""
        url = self._build_url(endpoint)
        
        # 设置默认超时
        kwargs.setdefault('timeout', self.timeout)
        
        # 记录请求
        self._log_request(method, url, **kwargs)
        
        start_time = time.time()
        try:
            response = self.session.request(method, url, **kwargs)
            self._log_response(response, start_time)
            return response
        except requests.exceptions.RequestException as e:
            print(f"[ERROR] 请求失败: {e}")
            raise
    
    def get(self, endpoint: str, **kwargs) -> requests.Response:
        """GET请求"""
        return self.request('GET', endpoint, **kwargs)
    
    def post(self, endpoint: str, **kwargs) -> requests.Response:
        """POST请求"""
        return self.request('POST', endpoint, **kwargs)
    
    def put(self, endpoint: str, **kwargs) -> requests.Response:
        """PUT请求"""
        return self.request('PUT', endpoint, **kwargs)
    
    def delete(self, endpoint: str, **kwargs) -> requests.Response:
        """DELETE请求"""
        return self.request('DELETE', endpoint, **kwargs)
    
    def close(self):
        """关闭session"""
        self.session.close()

2. 认证客户端

class AuthenticatedHTTPClient(BaseHTTPClient):
    """带认证的HTTP客户端"""
    
    def __init__(self, base_url: str, username: str = None, password: str = None, token: str = None):
        super().__init__(base_url)
        self.username = username
        self.password = password
        self.token = token
        
        if token:
            self.set_token(token)
        elif username and password:
            self.login(username, password)
    
    def set_token(self, token: str):
        """设置认证token"""
        self.token = token
        self.session.headers.update({'Authorization': f'Bearer {token}'})
    
    def login(self, username: str, password: str) -> bool:
        """用户登录"""
        login_data = {
            'username': username,
            'password': password
        }
        
        try:
            response = self.post('/auth/login', json=login_data)
            if response.status_code == 200:
                token = response.json().get('token')
                if token:
                    self.set_token(token)
                    return True
            return False
        except Exception as e:
            print(f"登录失败: {e}")
            return False
    
    def refresh_token(self) -> bool:
        """刷新token"""
        try:
            response = self.post('/auth/refresh')
            if response.status_code == 200:
                new_token = response.json().get('token')
                if new_token:
                    self.set_token(new_token)
                    return True
            return False
        except Exception as e:
            print(f"刷新token失败: {e}")
            return False

3. 业务API客户端

class UserAPIClient(AuthenticatedHTTPClient):
    """用户API客户端"""
    
    def get_user_profile(self, user_id: int) -> Dict[str, Any]:
        """获取用户资料"""
        response = self.get(f'/users/{user_id}')
        response.raise_for_status()
        return response.json()
    
    def create_user(self, user_data: Dict[str, Any]) -> Dict[str, Any]:
        """创建用户"""
        response = self.post('/users', json=user_data)
        response.raise_for_status()
        return response.json()
    
    def update_user(self, user_id: int, user_data: Dict[str, Any]) -> Dict[str, Any]:
        """更新用户"""
        response = self.put(f'/users/{user_id}', json=user_data)
        response.raise_for_status()
        return response.json()
    
    def delete_user(self, user_id: int) -> bool:
        """删除用户"""
        response = self.delete(f'/users/{user_id}')
        return response.status_code == 204
    
    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('/users/search', params=params)
        response.raise_for_status()
        return response.json()

4. 响应断言增强

class ResponseAssertion:
    """响应断言工具"""
    
    def __init__(self, response: requests.Response):
        self.response = response
    
    def assert_status_code(self, expected_code: int):
        """断言状态码"""
        actual_code = self.response.status_code
        assert actual_code == expected_code, f"期望状态码{expected_code}，实际{actual_code}"
        return self
    
    def assert_json_path(self, json_path: str, expected_value: Any):
        """断言JSON路径值"""
        import jsonpath
        
        json_data = self.response.json()
        actual_values = jsonpath.jsonpath(json_data, json_path)
        
        assert actual_values, f"JSON路径 {json_path} 未找到"
        actual_value = actual_values[0]
        assert actual_value == expected_value, f"路径 {json_path} 期望值 {expected_value}，实际值 {actual_value}"
        return self
    
    def assert_response_time(self, max_seconds: float):
        """断言响应时间"""
        actual_time = self.response.elapsed.total_seconds()
        assert actual_time <= max_seconds, f"响应时间 {actual_time}s 超过限制 {max_seconds}s"
        return self
    
    def assert_contains(self, text: str):
        """断言响应包含指定文本"""
        response_text = self.response.text
        assert text in response_text, f"响应中未找到文本: {text}"
        return self

# 使用示例
def test_user_api():
    client = UserAPIClient('https://api.example.com', 'admin', 'password')
    
    # 创建用户
    user_data = {'name': '张三', 'email': 'zhangsan@example.com'}
    response = client.post('/users', json=user_data)
    
    # 链式断言
    (ResponseAssertion(response)
     .assert_status_code(201)
     .assert_json_path('$.data.name', '张三')
     .assert_json_path('$.data.email', 'zhangsan@example.com')
     .assert_response_time(2.0))

实用技巧与最佳实践

1. 环境配置管理

import os
from dataclasses import dataclass

@dataclass
class APIConfig:
    """API配置类"""
    base_url: str
    username: str
    password: str
    timeout: int = 30
    verify_ssl: bool = True

class ConfigManager:
    """配置管理器"""
    
    @staticmethod
    def get_config(env: str = None) -> APIConfig:
        """获取环境配置"""
        env = env or os.getenv('TEST_ENV', 'dev')
        
        configs = {
            'dev': APIConfig(
                base_url='https://dev-api.example.com',
                username='dev_user',
                password='dev_pass'
            ),
            'test': APIConfig(
                base_url='https://test-api.example.com',
                username='test_user',
                password='test_pass'
            ),
            'prod': APIConfig(
                base_url='https://api.example.com',
                username='prod_user',
                password='prod_pass',
                verify_ssl=True
            )
        }
        
        return configs.get(env, configs['dev'])

# 使用配置
config = ConfigManager.get_config()
client = UserAPIClient(config.base_url, config.username, config.password)

2. 请求缓存

import hashlib
import json
import time
from typing import Dict, Any

class CachedHTTPClient(BaseHTTPClient):
    """带缓存的HTTP客户端"""
    
    def __init__(self, base_url: str, cache_ttl: int = 300):
        super().__init__(base_url)
        self.cache = {}
        self.cache_ttl = cache_ttl
    
    def _get_cache_key(self, method: str, url: str, **kwargs) -> str:
        """生成缓存键"""
        cache_data = {
            'method': method,
            'url': url,
            'params': kwargs.get('params'),
            'json': kwargs.get('json'),
            'data': kwargs.get('data')
        }
        cache_str = json.dumps(cache_data, sort_keys=True)
        return hashlib.md5(cache_str.encode()).hexdigest()
    
    def request(self, method: str, endpoint: str, use_cache: bool = True, **kwargs):
        """带缓存的请求方法"""
        if not use_cache or method.upper() != 'GET':
            return super().request(method, endpoint, **kwargs)
        
        url = self._build_url(endpoint)
        cache_key = self._get_cache_key(method, url, **kwargs)
        
        # 检查缓存
        if cache_key in self.cache:
            cached_data, timestamp = self.cache[cache_key]
            if time.time() - timestamp < self.cache_ttl:
                print(f"[CACHE HIT] {method.upper()} {url}")
                return cached_data
        
        # 发送请求并缓存结果
        response = super().request(method, endpoint, **kwargs)
        self.cache[cache_key] = (response, time.time())
        
        return response

总结

requests库虽然使用简单，但要用好它需要掌握很多技巧。通过合理的封装，我们可以构建出功能强大、易于维护的HTTP客户端。

关键要点回顾：

1. Session对象：保持连接状态，提高性能
2. 重试机制：处理网络不稳定问题
3. 合理封装：提高代码复用性和可维护性
4. 断言增强：让测试验证更加便捷
5. 配置管理：支持多环境切换

下一篇文章，我们将探讨测试数据管理与驱动策略，敬请期待！

推荐阅读

• pytest测试框架基础与实战
• 测试数据管理与驱动策略
• 接口自动化框架架构设计