时间戳

首页 > 解决方案 > 如果请求因代码失败，请重试类方法

python - 如果请求因代码失败，请重试类方法

问题描述

我有一个 Python 类，它使用boto3和requests库将有效负载发送到 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 调用会以 status 失败504。我想围绕这个类方法编写一个包装器重试方法，默认情况下会retry^2在每次尝试之间等待 5 次重试，如果代码成功则退出循环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。任何帮助，将不胜感激！

标签： pythonamazon-web-servicespython-requestsboto3

解决方案

我会尝试这样的事情：

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

当然，您可能希望根据自己的需要对其进行调整。

推荐阅读