如果请求失败,则重试类方法

2024-03-29 08:48:56 发布

您现在位置:Python中文网/ 问答频道 /正文

我有一个Python类,它用boto3requests库向AWS发送有效负载。但是,有时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。我想围绕这个类方法编写一个wrapper retry方法,默认情况下,在每次尝试之间重试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时返回它。任何帮助都将不胜感激!你知道吗


Tags: 方法代码fromimportselfauthresponsesession
1条回答
网友
1楼 · 发布于 2024-03-29 08:48:56

我会尝试这样的方法:

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('')



当然,你可以根据自己的需要来调整它。你知道吗

相关问题 更多 >