如何在Python requests中合并超时重试与PKCS12证书适配器?
合并超时重试与PKCS12证书适配器
问题背景
需要调用可靠性欠佳的Web端点,同时要实现两个需求:
- 应用超时/重试策略保障请求可靠性
- 使用PKCS12客户端证书完成身份认证
现有两个独立工作的适配器:TimeoutAdapter(处理超时和重试逻辑)和Pkcs12Adapter(处理PKI证书认证),需将二者合并为一个适配器,让requests.Session同时具备两种能力。
解决方案
创建一个继承自Pkcs12Adapter的新适配器类,整合超时逻辑并支持重试策略配置,既保留PKCS12证书的处理能力,又加入超时和重试机制。
合并后的完整代码
from requests_pkcs12 import Pkcs12Adapter from urllib3.util.retry import Retry DEFAULT_TIMEOUT = 30 class PkiTimeoutRetryAdapter(Pkcs12Adapter): def __init__(self, *args, **kwargs): # 提取超时参数,未指定则使用默认值 self.timeout = kwargs.pop("timeout", DEFAULT_TIMEOUT) # 调用父类初始化方法,传入PKCS12相关参数 super().__init__(*args, **kwargs) def send(self, request, **kwargs): # 若请求未单独设置超时,使用类实例的超时配置 timeout = kwargs.get("timeout") if timeout is None: kwargs["timeout"] = self.timeout # 调用父类send方法,执行证书认证并发送请求 return super().send(request, **kwargs) # 会话配置与使用示例 if __name__ == "__main__": import requests target_url = 'https://some-webservice.com' session = requests.Session() # 配置重试策略 retry_strategy = Retry( total=10, status_forcelist=[429, 500, 502, 503, 504], backoff_factor=1 ) # 初始化合并后的适配器,传入所有必要参数 pki_retry_adapter = PkiTimeoutRetryAdapter( pkcs12_filename='path/to/cert.p12', pkcs12_password='cert-password', timeout=30, # 可选,不填则使用DEFAULT_TIMEOUT max_retries=retry_strategy ) # 将适配器挂载到目标URL session.mount(target_url, pki_retry_adapter) # 发送请求,自动应用超时、重试和PKCS12认证 response = session.get(f"{target_url}/your-endpoint") print(response.status_code)
代码说明
- 继承逻辑:新类
PkiTimeoutRetryAdapter继承自Pkcs12Adapter,天然具备PKCS12证书的解析与认证能力。 - 超时处理:在初始化阶段提取超时参数,发送请求时自动填充未指定超时的请求。
- 重试机制:通过
max_retries参数传入Retry对象,借助Pkcs12Adapter的父类HTTPAdapter的重试逻辑实现自动重试。 - 扩展性:保留原两个适配器的所有可配置参数,可根据需求调整超时时间、重试次数、证书路径等。
验证效果
运行代码后,发送的请求会自动完成:
- 使用指定的PKCS12证书进行身份认证
- 遵循预设的超时时间限制
- 遇到
429/5xx状态码时自动重试,并按配置的退避策略等待
内容的提问来源于stack exchange,提问作者auslander
相关产品推荐
相关产品推荐

