1. 为什么选择requests模块做接口自动化
在Python生态中,处理HTTP请求的库至少有十几种选择,但requests能成为事实上的标准库绝非偶然。我2016年第一次用urllib2做爬虫时,光是处理一个带cookie的POST请求就写了20多行代码,而用requests只需要3行。这种开发效率的差距,在接口自动化测试这种需要高频发送请求的场景下会被放大十倍不止。
requests的核心优势在于:
- 人性化的API设计:
response = requests.get(url)这种近乎自然语言的写法,让代码可读性大幅提升 - 完善的会话管理:自动处理cookies、连接池等底层细节,不像urllib需要手动维护
- 丰富的功能支持:从基础认证到文件上传,覆盖了90%的HTTP使用场景
- 活跃的社区生态:GitHub上48k+的star和8k+的fork意味着遇到问题随时能找到解决方案
提示:虽然Python标准库有urllib3,但在实际接口测试中,除非有特殊需求(如需要精细控制HTTP连接池),否则直接使用requests是更明智的选择。
2. GET请求的完整参数解析
2.1 基础GET请求实现
先看最简单的GET请求示例:
import requests response = requests.get('https://api.github.com/events') print(response.status_code) # 200 print(response.text[:100]) # 查看前100个字符这个例子虽然简单,但隐藏了几个关键知识点:
- 自动解码:response.text会根据响应头自动处理编码(如UTF-8)
- 连接复用:底层使用urllib3的连接池,相同host的请求会复用TCP连接
- 超时机制:默认没有设置超时,这在生产环境是危险的(后面会讲如何正确设置)
2.2 带参数的GET请求
实际接口测试中,90%的GET请求都需要携带参数。requests提供了两种传参方式:
方式一:手动拼接URL(不推荐)
url = "https://httpbin.org/get?key1=value1&key2=value2" response = requests.get(url)方式二:使用params参数(推荐)
params = {'key1': 'value1', 'key2': ['value2', 'value3']} response = requests.get('https://httpbin.org/get', params=params)第二种方式的优势在于:
- 自动处理URL编码(比如空格转%20)
- 支持列表参数(如上例中的value2和value3会变成
key2=value2&key2=value3) - 代码可读性更好,参数与URL分离
2.3 请求头定制
接口测试经常需要设置User-Agent、Content-Type等头部信息:
headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)', 'Accept': 'application/json' } response = requests.get('https://httpbin.org/headers', headers=headers)实测中发现三个易错点:
- Header名称大小写:HTTP协议规定header名不区分大小写,但某些服务端实现会有要求
- 默认User-Agent:requests默认的UA是
python-requests/2.31.0,容易被某些网站屏蔽 - Content-Type陷阱:GET请求通常不需要设置,但某些特殊API可能有要求
3. 实战中的高级技巧
3.1 超时控制与重试机制
我曾在生产环境遇到因未设置超时导致线程挂起的严重事故。正确的超时设置应该是:
try: response = requests.get('https://httpbin.org/delay/5', timeout=(3.05, 27)) except requests.exceptions.Timeout: print("请求超时")这里的timeout参数有两个值:
- 第一个3.05秒是连接超时(建议比3的整数倍稍大,避免TCP重传定时器干扰)
- 第二个27秒是读取超时(根据业务响应时间合理设置)
对于429 Too Many Requests错误(常见于爬虫场景),可以这样实现自动重试:
from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session = requests.Session() retries = Retry(total=3, backoff_factor=1, status_forcelist=[429, 502, 503, 504]) session.mount('https://', HTTPAdapter(max_retries=retries)) response = session.get('https://httpbin.org/status/429')3.2 代理配置与SSL验证
在企业内网测试时经常需要配置代理:
proxies = { 'http': 'http://10.10.1.10:3128', 'https': 'http://10.10.1.10:1080', } requests.get('http://example.org', proxies=proxies)遇到自签名证书报错时(常见于测试环境),可以这样处理:
# 临时关闭验证(不安全,仅限测试环境) requests.get('https://self-signed.badssl.com', verify=False) # 添加自定义CA证书 requests.get('https://self-signed.badssl.com', verify='/path/to/cert.pem')3.3 响应处理最佳实践
很多新手会直接使用response.text,这其实存在隐患。更健壮的做法是:
response = requests.get('https://api.github.com/events') # 检查状态码 response.raise_for_status() # 非2xx会抛出异常 # 智能解码内容(优先用response.json()处理JSON数据) try: data = response.json() except ValueError: data = response.content # 二进制数据 # 或者 response.text # 文本数据我总结的响应处理黄金法则:
- 永远先检查状态码(或调用raise_for_status)
- JSON响应优先用response.json()解析
- 大文件下载用response.iter_content()分块读取
4. 企业级接口测试实战
4.1 封装可复用的请求工具类
在实际项目中,我会这样封装GET请求:
class ApiClient: def __init__(self, base_url): self.session = requests.Session() self.base_url = base_url self.session.headers.update({'User-Agent': 'MyApiClient/1.0'}) def get(self, endpoint, params=None, **kwargs): url = f"{self.base_url}/{endpoint}" try: response = self.session.get( url, params=params, timeout=10, **kwargs ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"请求失败: {e}") raise # 使用示例 client = ApiClient('https://api.example.com/v1') data = client.get('users', params={'active': True})这种封装带来了三个好处:
- 统一管理基础URL和公共headers
- 内置异常处理和日志记录
- 一致的超时和重试策略
4.2 结合pytest做自动化测试
实际项目中GET请求测试可以这样写:
import pytest @pytest.mark.parametrize("user_id,expected_status", [ (1, 200), (999, 404), ("invalid", 400) ]) def test_get_user(user_id, expected_status): response = requests.get(f"https://api.example.com/users/{user_id}") assert response.status_code == expected_status if response.status_code == 200: assert "username" in response.json()4.3 性能优化技巧
当需要批量发送GET请求时,同步请求效率很低。这时可以用:
import concurrent.futures urls = [ 'https://httpbin.org/get?page=1', 'https://httpbin.org/get?page=2' ] def fetch(url): return requests.get(url).json() with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: results = list(executor.map(fetch, urls))但要注意:
- 线程数不要超过目标服务器的承受能力
- 考虑使用timeout参数避免单个请求拖慢整体
- 对于特别大量的请求,建议使用aiohttp等异步库
5. 常见问题排查手册
5.1 SSL证书验证失败
错误信息:SSLError: HTTPSConnectionPool...certificate verify failed
解决方案:
# 临时方案(测试环境) requests.get(url, verify=False) # 长期方案 import certifi requests.get(url, verify=certifi.where())5.2 编码问题导致乱码
现象:response.text返回乱码
正确处理方式:
# 方式1:手动指定编码 response.encoding = 'gbk' print(response.text) # 方式2:使用二进制内容自行解码 content = response.content print(content.decode('gb18030'))5.3 连接池耗尽问题
现象:ConnectionError: HTTPSConnectionPool...Max retries exceeded
解决方案:
# 增加连接池大小 from requests.adapters import HTTPAdapter session = requests.Session() adapter = HTTPAdapter(pool_connections=50, pool_maxsize=50) session.mount('https://', adapter)5.4 处理gzip压缩响应
有些服务器会返回gzip压缩内容,requests默认会自动解压,但偶尔会遇到问题:
# 手动处理gzip import gzip from io import BytesIO if response.headers.get('Content-Encoding') == 'gzip': buf = BytesIO(response.content) decompressed = gzip.GzipFile(fileobj=buf).read() print(decompressed.decode())