如何在 Python 请求中组合多个适配器？

Question

我需要对一个不太可靠的 Web 端点进行多次调用，因此我整理了一个超时/重试策略，并将其作为适配器发送给

requests.Session()

对象。

但是，我还需要使用客户端 PKCS12 证书和公共证书颁发机构（验证）来安装相同的端点，我可以使用

requests_pkcs12

包轻松执行此操作 - 这也会创建一个适配器。

所以，适配器 1 看起来像这样，（与我在互联网上找到的东西拼凑在一起）：

# timeout.py

from requests.adapters import HTTPAdapter

DEFAULT_TIMEOUT = 30

class TimeoutAdapter(HTTPAdapter):
    def __init__(self, *args, **kwargs):
        self.timeout = DEFAULT_TIMEOUT
        if "timeout" in kwargs:
            self.timeout = kwargs["timeout"]
        super().__init__(*args, **kwargs)
    
    def send(self, request, **kwargs):
        timeout = kwargs.get("timeout")
        if timeout is None:
            kwargs["timeout"] = self.timeout
        return super().send(request, **kwargs)

## set up adapter
if __name__ == "__main__":
    
    from urllib3.util.retry import Retry
    
    u = 'https://some-webservice.com'
    s = requests.Session()
    retry_strategy = Retry(
        total=10,
        status_forcelist=[429, 500, 502, 503, 504],
        backoff_factor=1)
    s.mount(u, adapter=TimeoutAdapter(max_retries=retry_strategy)

然后，第二个适配器如下所示：

from requests import Session
from requests_pkcs12 import Pkcs12Adapter

pki_adapter = Pkcs12Adapter(
    pkcs12_filename='path/to/cert.p12',
    pkcs12_password='cert-password')

u = 'https://some-webservice.com'
s = requests.Session()
s.mount(u, adapter=pki_adapter)

两者都可以很好地工作，但是我如何将它们合并到一个适配器中，以便会话使用 PKI 进行身份验证，同时也遵循超时/重试策略？

Answer 1

你有没有想过做这样的事情？

from requests.adapters import HTTPAdapter
from requests_pkcs12 import Pkcs12Adapter
from urllib3.util.retry import Retry

DEFAULT_TIMEOUT = 30

class TimeoutPkcs12Adapter(TimeoutAdapter, Pkcs12Adapter):
    def __init__(self, *args, **kwargs):
        self.timeout = DEFAULT_TIMEOUT
        if "timeout" in kwargs:
            self.timeout = kwargs["timeout"]
        self.pkcs12_filename = kwargs.pop("pkcs12_filename", None)
        self.pkcs12_password = kwargs.pop("pkcs12_password", None)

        HTTPAdapter.__init__(self, *args, **kwargs)
        Pkcs12Adapter.__init__(self, pkcs12_filename=self.pkcs12_filename, pkcs12_password=self.pkcs12_password)

    def send(self, request, **kwargs):
        timeout = kwargs.get("timeout")
        if timeout is None:
            kwargs["timeout"] = self.timeout
        return super().send(request, **kwargs)

if __name__ == "__main__":
    u = 'https://some-webservice.com'
    s = requests.Session()
    retry_strategy = Retry(
        total=10,
        status_forcelist=[429, 500, 502, 503, 504],
        backoff_factor=1
    )
    s.mount(
        u,
        adapter=TimeoutPkcs12Adapter(
            max_retries=retry_strategy,
            pkcs12_filename='path/to/cert.p12',
            pkcs12_password='cert-password'
        )
    )

它将帮助您通过继承组合到适配器。