Как я могу объединить несколько адаптеров в запросах Python?
Мне нужно сделать несколько вызовов к конечной веб-точке, что несколько ненадежно, поэтому я разработал стратегию тайм-аута/повторной попытки, которую я передаю объекту 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 для аутентификации, но также соблюдал стратегию тайм-аута/повторной попытки?
Я не могу проверить это в данный момент, но поскольку requests_pkcs12.Pkcs12Adapter наследует от requests.adapters.HTTPAdapter, я смогу просто настроить retry_strategy как обычно и сделать: s.mount(u, adapter=Pkcs12Adapter(cert, pw, verify, max_retries=retry_strategy), верно?
Ответы 2
Думали ли вы сделать что-то подобное?
from requests.adapters import HTTPAdapter
from requests_pkcs12 import Pkcs12Adapter
from urllib3.util.retry import Retry
import requests
import logging
DEFAULT_TIMEOUT = 30
class TimeoutPkcs12Adapter(HTTPAdapter):
def __init__(self, pkcs12_filename, pkcs12_password, timeout=DEFAULT_TIMEOUT, **kwargs):
self.timeout = timeout
self.pkcs12_adapter = Pkcs12Adapter(pkcs12_filename=pkcs12_filename, pkcs12_password=pkcs12_password)
super().__init__(**kwargs)
def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
# Load certificate details into pool manager
pool_kwargs['cert_file'] = self.pkcs12_adapter.cert_file
pool_kwargs['key_file'] = self.pkcs12_adapter.key_file
super().init_poolmanager(connections, maxsize, block, **pool_kwargs)
def send(self, request, **kwargs):
kwargs.setdefault('timeout', self.timeout)
try:
return super().send(request, **kwargs)
except requests.exceptions.RequestException as e:
logging.error(f"Request failed: {e}")
raise
if __name__ == "__main__":
url = 'https://some-webservice.com'
session = requests.Session()
retry_strategy = Retry(total=10, status_forcelist=[429, 500, 502, 503, 504], backoff_factor=1)
adapter = TimeoutPkcs12Adapter(
pkcs12_filename='path/to/cert.p12',
pkcs12_password='cert-password',
max_retries=retry_strategy
)
session.mount(url, adapter)
это поможет вам объединиться с адаптерами по наследству.
Обновление: в предыдущем коде была проблема с Diamond.
Спасибо! Но вы имели в виду class TimeoutPkcs12Adapter(HTTPAdapter, Pkcs12Adapter):?
Да, извините за ошибку, он должен наследовать от HTTPAdapter
К сожалению, это дает мне Cannot create consistent method ordering в классе. наследование — это сложно
попробовал локально, и это сработало, в предыдущем ответе была проблема с алмазами.
Еще раз спасибо. Что в данном случае представляет собой файл ключа? Пакеты PKCS12 расшифровываются в памяти и не имеют файлов-ключей, так что, думаю, можно было бы просто пропустить это? У меня есть файл CA (.crt), на который Session().verify будет указывать, но для клиентов все находится в одном файле .p12.
Я не уверен, что вы сможете сделать это правильно без key_file, но, пожалуйста, попробуйте и дайте мне знать, я тоже могу чему-то научиться из этого, спасибо <3
В итоге я отметил ответ @booboo, хотя он был очень близок к вашему, порядок может быть полезен и вам. Я очень благодарен вам за ваше время и ответы, они все равно были полезны!
Предполагая, что вам нужно поведение тайм-аута по умолчанию от TimeoutAdapter, тогда, поскольку TimeoutAdapter и Pkcs12Adapter оба наследуют от HTTPAdapter, вы можете наследовать первое от второго (одиночное наследование). Например:
# timeout.py
from requests import Session
from requests_pkcs12 import Pkcs12Adapter
DEFAULT_TIMEOUT = 30
class TimeoutAdapter(Pkcs12Adapter):
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
retry_strategy = Retry(
total=10,
status_forcelist=[429, 500, 502, 503, 504],
backoff_factor=1)
timeout_adapter = TimeoutAdapter(
max_retries=retry_strategy,
pkcs12_filename='path/to/cert.p12',
pkcs12_password='cert-password'
)
u = 'https://some-webservice.com'
s = requests.Session()
s.mount(u, adapter=timeout_adapter)
Именно то, что я ищу. Дает возможность предлагать его в качестве дополнительного класса соединения, если в определенных случаях это не требуется. Спасибо!
Отвечает ли это на ваш вопрос? Запросы Python: могу ли я использовать s.mount() последовательно?