python - Python中的重试函数

Question

前段时间，我需要retryR 中的一个函数来处理缓慢的服务器响应。该函数将具有以下行为：（尝试一个操作（函数或方法），如果失败，请稍等然后重试）x10

我想出了以下内容：

retry <- function(fun, max_trys = 10, init = 0){
  suppressWarnings(tryCatch({
    Sys.sleep(0.3);
    if(init<max_trys) {fun}
}, error=function(e){retry(fun, max_trys, init = init+1)}))}

它运作良好。现在我在 Python3 中需要相同的，所以我尝试制作相同的代码：

import time
def retry_fun(fun, max_trys = 10, init=0):
    try:
        time.sleep(0.3)
        if(init<max_trys):
            fun
    except:
        retry_fun(fun, max_trys, init = init+1)

但是当我运行它时，它会使我的内核崩溃。由于我是 Python 的初学者，我不确定导致崩溃的原因，以及是否/如何将函数作为参数传递给另一个函数。

你能帮帮我吗？

Answer 1

除了能够传递函数并通过()在名称后添加（Python 调用调用的语法）来使用它们，您不需要使用递归；把它放在一个循环中：

import time
def retry(fun, max_tries=10):
    for i in range(max_tries):
        try:
           time.sleep(0.3) 
           fun()
           break
        except Exception:
            continue

except Exception应该更改以捕获该函数可能引发的有意义的异常。使用Exception（就像我在示例中所做的那样）通常是不好的做法，因为它会捕获一大类您可能不想捕获的异常。

除此之外，使用 afor-loop而不是显式的第三个计数器和递归（这会导致大值的长调用堆栈）更好。

Answer 2

我知道这是一个老问题。但是，我想添加我已经准备好的解决方案。最好的方法是编写一个retry装饰器，当异常发生时会重试。此外，您还可以设置自定义指数延迟。文档字符串解释了如何使用装饰器。干得好：

import logging
import time
from functools import partial, wraps


def retry(func=None, exception=Exception, n_tries=5, delay=5, backoff=1, logger=False):
    """Retry decorator with exponential backoff.

    Parameters
    ----------
    func : typing.Callable, optional
        Callable on which the decorator is applied, by default None
    exception : Exception or tuple of Exceptions, optional
        Exception(s) that invoke retry, by default Exception
    n_tries : int, optional
        Number of tries before giving up, by default 5
    delay : int, optional
        Initial delay between retries in seconds, by default 5
    backoff : int, optional
        Backoff multiplier e.g. value of 2 will double the delay, by default 1
    logger : bool, optional
        Option to log or print, by default False

    Returns
    -------
    typing.Callable
        Decorated callable that calls itself when exception(s) occur.

    Examples
    --------
    >>> import random
    >>> @retry(exception=Exception, n_tries=4)
    ... def test_random(text):
    ...    x = random.random()
    ...    if x < 0.5:
    ...        raise Exception("Fail")
    ...    else:
    ...        print("Success: ", text)
    >>> test_random("It works!")
    """

    if func is None:
        return partial(
            retry,
            exception=exception,
            n_tries=n_tries,
            delay=delay,
            backoff=backoff,
            logger=logger,
        )

    @wraps(func)
    def wrapper(*args, **kwargs):
        ntries, ndelay = n_tries, delay

        while ntries > 1:
            try:
                return func(*args, **kwargs)
            except exception as e:
                msg = f"{str(e)}, Retrying in {ndelay} seconds..."
                if logger:
                    logging.warning(msg)
                else:
                    print(msg)
                time.sleep(ndelay)
                ntries -= 1
                ndelay *= backoff

        return func(*args, **kwargs)

    return wrapper

Answer 3

有几个 Python 包：

1. 退避
2. 韧性

退避示例

import backoff

@backoff.on_exception(backoff.expo,
                      (MyPossibleException1,
                       MyPossibleException2))
def your_function(param1, param2):
    # Do something

坚韧的例子

from tenacity import wait_exponential, retry, stop_after_attempt

@retry(wait=wait_exponential(multiplier=2, min=2, max=30),  stop=stop_after_attempt(5))
def your_function(param1, param2):
   # Do something

Answer 4

您可以执行以下操作：

def retry_call(function_name, args=None, kwargs=None, retries=3):
    pass_on_args = args if args else []
    pass_on_kwargs = kwargs if kwargs else {}
    for index in range(1, retries+1):
        try:
            return function_name(*pass_on_args, **pass_on_kwargs)
        except Exception as error:
            if index == retries:
                LOGGER.error("Failed %s attempts at calling: %s", retries, function_name)
                LOGGER.error("    args: %s", args)
                LOGGER.error("    kwargs: %s", kwargs)
                LOGGER.error("    error: %s", error)
                raise
            else:
                LOGGER.warning("Failed %d attempt(s) at calling: %s", index, function_name)
                LOGGER.warning("Retrying")

def get_variable
   with open(FILE) as info:
        for line in info:
            if 'Mode' in line:
            return mode

然后你会调用这个函数：

mode = retry(get_variable)
print(mode)

Answer 5

还有另一个 pip 可安装重试包： https ://pypi.org/project/retry/

pip install retry