Повторите метод класса, если запросы не с кодом - PullRequest
Новая
       12

Повторите метод класса, если запросы не с кодом

0 голосов
/ 24 сентября 2019

У меня есть класс Python, который отправляет полезные данные в AWS с библиотекой boto3 и requests.Однако иногда http-запросы терпят неудачу с различными кодами, поэтому я хотел написать внутри класса функцию-обертку, которая будет повторять попытку отправки полезной нагрузки 5 раз, если он получит определенные коды, и вызовет исключение, если он полностью потерпит неудачу.Вот метод класса (предположим, что вызовы методов работают как положено): 

import requests
from boto3 import Session

def update_status(self, status):
    payload = status

    auth = self.sign_request()
    response = requests.patch(self.url, auth=auth, data=payload)

    status_code = response.status_code
    response_text = response.text

    if not response.ok:
        logging.error("Failed updating status of request: " + str(
            {'host': self.host, 'region': self.region,
             'service': self.service, 'url': self.url, 'status': str(status)}))

        raise IOError('Update training status failed with status code: ' + str(status_code) + '\n' + response_text)

    logging.info("Updated status")

Иногда этот вызов API завершается с ошибкой со статусом 504.Я хотел бы написать метод повтора оболочки вокруг этого метода класса, который по умолчанию будет повторять 5 раз с ожиданием retry^2 между каждой попыткой и выходить из цикла, если это успешно с кодом 200.

Я нашел этот код, который, кажется, соответствует тому, что я использовал бы, я просто не уверен, как обернуть мой текущий метод в него и вызвать его: 

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


def requests_retry_session(
    retries=5,
    backoff_factor=0.3,
    status_forcelist=(500, 502, 504),
    session=None,
):
    session = session or requests.Session()
    retry = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=backoff_factor,
        status_forcelist=status_forcelist,
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    return session

Проблема с приведенным выше кодомон использует requests.session и возвращает его, пока мой класс уже использует boto3.Session.Любая помощь будет оценена!

1 Ответ

1 голос
/ 25 сентября 2019

Я бы попробовал что-то вроде этого: 

import time
import requests
from functools import wraps
import logging

logging.basicConfig(level=logging.DEBUG)

def retry(delay=10, retries=4):
    def retry_decorator(f):
        @wraps(f)
        def f_retry(*args, **kwargs):
            opt_dict = {'retries': retries, 'delay': delay}
            while opt_dict['retries'] > 1:
                try:
                    return f(*args, **kwargs)
                except Exception as e:
                    msg = "Exception: {}, Retrying in {} seconds...".format(e, delay)
                    print(msg)
                    time.sleep(opt_dict['delay'])
                    opt_dict['retries'] -= 1
            return f(*args, **kwargs)

        return f_retry

    return retry_decorator


class YourClass:
    # JUST MOCK FOR PROOF OF CONCEPT
    url = 'YOUR URL'
    status = 'YOUR STATUS'
    def sign_request(self):
        return ''
    host = 'YOUR HOST'
    region = 'YOUR REGION'
    service = 'YOUR SERVICE'
    # MOCK END

    def update_status(self, status):
        payload = status

        auth = self.sign_request()

        @retry(1, 5)
        def get_status():
            response = requests.patch(self.url, auth=auth, data=payload)

            if not response.ok:
                logging.error("Failed updating status of request: " + str(
                    {'host': self.host, 'region': self.region,
                     'service': self.service, 'url': self.url, 'status': str(status)}))

                raise IOError('Update training status failed with status code: ' + str(response.status_code) + '\n' + response.text)
            return response

        res = get_status()
        status_code = res.status_code
        response_text = res.text

        logging.info("Updated status")


x = YourClass()
x.url = 'https://httpstat.us/200'
x.update_status('')
x.url = 'https://httpstat.us/504'
x.update_status('')

Конечно, вы можете настроить его под свои нужды.

...