使用 Python 装饰器重试请求

Question

我的脚本中有多个执行 REST API api 请求的函数。由于我需要处理错误场景，因此我设置了如下重试机制。

no_of_retries = 3
def check_status():
    for i in range(0,no_of_retries):
        url = "http://something/something"
        try:
            result = requests.get(url, auth=HTTPBasicAuth(COMMON_USERNAME, COMMON_PASSWORD)).json()
            if 'error' not in result:
                return result
            else:
                continue
        except Exception as e:
            continue
    return None

我有几种不同的方法可以进行类似的操作。我们如何才能更好地避免重复可能是使用装饰器。

Answer 1

如果您不介意安装库，可以使用tenacity (github.com/jd/tenacity) 模块。他们的例子之一：

import random
from tenacity import retry, stop_after_attempt

# @retry  # retry forever
@retry(stop=stop_after_attempt(3))
def do_something_unreliable():
    if random.randint(0, 10) > 1:
        raise IOError("Broken sauce, everything is hosed!!!111one")
    else:
        return "Awesome sauce!"

print(do_something_unreliable())

这还允许您指定要继续重试的尝试次数或秒数。

对于您的情况，这可能看起来像这样（未经测试！）：

@retry(stop=stop_after_attempt(3))
def retry_get():
    result = requests.get(
            url, auth=HTTPBasicAuth(COMMON_USERNAME, COMMON_PASSWORD)).json()
    if 'error' not in result:
        raise RequestException(result)

Answer 2

您可以使用这样的装饰器并处理您自己的异常。

def retry(times, exceptions):
    """
    Retry Decorator
    Retries the wrapped function/method `times` times if the exceptions listed
    in ``exceptions`` are thrown
    :param times: The number of times to repeat the wrapped function/method
    :type times: Int
    :param Exceptions: Lists of exceptions that trigger a retry attempt
    :type Exceptions: Tuple of Exceptions
    """
    def decorator(func):
        def newfn(*args, **kwargs):
            attempt = 0
            while attempt < times:
                try:
                    return func(*args, **kwargs)
                except exceptions:
                    print(
                        'Exception thrown when attempting to run %s, attempt '
                        '%d of %d' % (func, attempt, times)
                    )
                    attempt += 1
            return func(*args, **kwargs)
        return newfn
    return decorator

@retry(times=3, exceptions=(ValueError, TypeError))
def foo1():
    print('Some code here ....')
    print('Oh no, we have exception')
    raise ValueError('Some error')

foo1()

Answer 3

第三方retry module 现在被广泛接受。您还可以传递要重试的异常列表、重试次数、延迟、最大延迟、指数回退等。

$ pip install retry

示例用法：

from retry import retry

@retry(ZeroDivisionError, tries=3, delay=2)
def make_trouble():
    '''Retry on ZeroDivisionError, raise error after 3 attempts, sleep 2 seconds between attempts.'''

Answer 4

除了使用装饰器，更好的解决方案可能是将请求移动到它自己的函数中，从而得到一个类似于这样的结构：

no_of_retries = 3

def make_request(url):
    for i in range(0,no_of_retries):
        try:
            result = requests.get(url, auth=HTTPBasicAuth(COMMON_USERNAME, COMMON_PASSWORD)).json()
            if 'error' not in result:
                return result
            else:
                continue
        except Exception as e:
            continue
    return result

def check_status():
    result = make_request("http://something/status")

def load_file():
    result = make_request("http://something/file")

这样，您可以在封装请求时避免重复代码。如果您要使用装饰器，则需要包装整个 load_file() 方法，这将阻止您在此函数中进一步处理请求的结果。

Answer 5

在 mrkiril 的回答之上使用 functools：

from functools import wraps, partial

def retry(f=None, times=10):
    if f is None:
        return partial(retry, times=times)

    @wraps(f)
    def wrap(*args, **kwargs):
        attempt = 0
        while attempt < times:
            try:
                return f(*args, **kwargs)
            except:
                print(f"{f.__name__}, attempt {attempt} of {times}")
                attempt += 1
        return f(*args, **kwargs)
    return wrap

然后，像下面这样包装你的函数：

import random

@retry
def foo():
    if random.randint(0, 5) != 0:
        raise Exception