使用Python装饰器重试请求

xzlaal3s  于 2023-06-28  发布在  Python
关注(0)|答案(8)|浏览(143)

我在我的脚本中有多个函数,它执行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

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

iq0todco

iq0todco1#

如果你不介意安装一个库,你可以使用tenacitygithub.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)
ibrsph3r

ibrsph3r2#

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

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

ttygqcqt3#

第三方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.'''
im9ewurl

im9ewurl4#

生产级示例

import logging
import time
import functools
import traceback

LOG_FORMAT = "%(asctime)s - %(levelname)s - %(pathname)s - %(funcName)s - %(lineno)d -msg: %(message)s"
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)

def retry(retry_num, retry_sleep_sec):
    """
    retry help decorator.
    :param retry_num: the retry num; retry sleep sec
    :return: decorator
    """
    def decorator(func):
        """decorator"""
        # preserve information about the original function, or the func name will be "wrapper" not "func"
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            """wrapper"""
            for attempt in range(retry_num):
                try:
                    return func(*args, **kwargs)  # should return the raw function's return value
                except Exception as err:   # pylint: disable=broad-except
                    logging.error(err)
                    logging.error(traceback.format_exc())
                    time.sleep(retry_sleep_sec)
                logging.error("Trying attempt %s of %s.", attempt + 1, retry_num)
            logging.error("func %s retry failed", func)
            raise Exception('Exceed max retry num: {} failed'.format(retry_num))

        return wrapper

    return decorator

用法

# this means try your function 5 times, each time sleep 60 seconds
@retry(5, 60)
def your_func():
    pass

正式参考:https://peps.python.org/pep-0318/

klr1opcd

klr1opcd5#

我推荐使用retry库,就像@MohitC提到的那样。但是,如果你像我一样被限制导入第三方库,欢迎你尝试我的版本:

import time

def retry(tries= -1, delay=0, max_delay=None, backoff=1, exceptions=Exception, log=False):
    """Retry Decorator with arguments
    Args:
        tries (int): The maximum number of attempts. Defaults to -1 (infinite)
        delay (int, optional): Delay between attempts (seconds). Defaults to 0
        max_delay (int, optional): The maximum value of delay (seconds). Defaults to None (Unlimited)
        backoff (int, optional): Multiplier applied to delay between attempts (seconds). Defaults to 1 (No backoff)
        exceptions (tuple, optional): Types of exceptions to catch. Defaults to Exception (all)
        log (bool, optional): Print debug logs. Defaults to False
    """
    def retry_decorator(func):
        def retry_wrapper(*args, **kwargs):
            nonlocal tries, delay, max_delay, backoff, exceptions, log
            while tries:
                try:
                    return func(*args, **kwargs)
                except exceptions:
                    tries -= 1

                    # Reached to maximum tries
                    if not tries:
                        raise
                    
                    # Log the retry logs for the given function
                    if log:
                        print(f"Retrying {func.__name__} in {delay} seconds")

                    # Apply delay between requests
                    time.sleep(delay)

                    # Adjust the next delay according to backoff
                    delay *= backoff

                    # Adjust maximum delay duration
                    if max_delay is not None:
                        delay = min(delay, max_delay)

        return retry_wrapper
    return retry_decorator

使用示例:

很简单:

@retry(10, delay=5)
    def do_something(params):
        # Example func to retry
        pass

高级:

@retry(10, delay=1, backoff=2, max_delay=10, exceptions=(TimeoutError), log=True)
    def do_something(params):
        # Example func to retry only for TimeoutErrors
        pass
mbjcgjjk

mbjcgjjk6#

与其使用装饰器,可能更好的解决方案是将请求移动到它自己的函数中,得到类似于下面的结构:

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")

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

wr98u20j

wr98u20j7#

我创建了一个自定义重试函数。如果第一个数字小于第二个数字,则该函数将重试。

验证码:

import time

# CUSTOM EXCEPTION
class custom_error(Exception):
    pass

# RETRY FUNCTION.
def retry(func, retries=3):
    print(func)

    def retry_wrapper(*args, **kwargs):
        print(args)
        n = args[0]
        u = args[1]
        print(n, u)

        attempts = 0
        while attempts < retries:
            try:
                if n > u:
                    return func(*args, **kwargs)
                else:
                    raise custom_error
            except custom_error:
                print("error")
                time.sleep(2)
                attempts += 1

    return retry_wrapper

@retry
def akash(a, b):
    c = a / b
    return c

# CALLING THE FUNCTION
a = akash(1, 2)
print(a)

输出:

<function akash at 0x00000187C3A66B00>
(1, 2)
1 2
error
error
error
vojdkbi0

vojdkbi08#

在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

然后,像下面这样 Package 你的函数:

import random

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

相关问题