Python代码重试功能

def is_user_exist(username):

    try:
        pwd.getpwnam(username)
        log.info("User %s exist" % username)
        return True
    except KeyError:
        log.exception("User %s does not exist." % username)
        return False

def retry(func, *func_args, **kwargs):
    retry_count = kwargs.get("retry_count", 5)
    delay = kwargs.get("delay", 5)
    while retry_count > 1:
        if func(*func_args):
            return
        log.debug("waiting for %s seconds before retyring again")
        sleep(delay)
        retry_count = retry_count - 1

    return func(*func_args)

#1 楼

我喜欢所有的Ev。 Kounis的回答，所以我将添加一些更高级的详细信息。

说实话-y

现在，您并不严格要求func返回True或False ，只不过是对或错。很好，并且是Pythonic。我认为您将受益于在文档字符串中指出这一点（您也应该编写）。另外，可能值得返回func(*func_args)的实际值，而不是True或False；这样，如果您想利用值的真实性，就可以从函数中实际获取值。

更好的kwargs支持

您也不要允许将任何关键字参数传递给函数-函数可能支持它们，因此您应该这样做。我会将您从kwargs中拉出的两个关键字参数提升为具有适当默认值（在您的情况下为5）的显式关键字参数。

例外情况函数确实具有在异常之后重试的任何概念。我不希望您这样做，因为我希望这是显而易见的原因

for _ in range(retry_count):
    try:
        if func(*func_args):
            return True
    except:
        pass
    log.debug("wating for %s seconds before retrying again")
    sleep delay)

但是，在很多情况下，我怀疑您会知道您可能会遇到哪些例外情况期望（例如，网络超时或连接性故障），并且您可能希望由重试框架处理。为此，我认为这样可能会很好：

def retry(func, *args, retry_count=5, delay=5, allowed_exceptions=(), **kwargs):
    for _ in range(retry_count):
        try:
            result = func(*args, **kwargs)
            if result: return result
        except allowed_exceptions as e:
            pass

这显然不是完整的实现；我遗漏了您拥有的其他部分，如果它在上一次迭代中失败，它的行为会很奇怪，但是应该足以启动。认为如果它是装饰者，我们还可以从中获得更多价值。这样，函数的使用者甚至不需要知道是否要重试该函数。他们只是调用它们的函数，然后看它是否起作用，并且是否重试就变得无关紧要了。不要忘记使用functools.wraps来保存元数据。

import functools

def retry(retry_count=5, delay=5, allowed_exceptions=()):
    def decorator(f):
        @functools.wraps(f)
        def wrapper(*args, **kwargs):
            for _ in range(retry_count):
                # everything in here would be the same

        return wrapper
    return decorator

然后您可以对所有人启用重试，如下所示：当你在等什么？使用asyncio可能会做更多的事情（这是针对Python 3.5及更高版本的）。在此之前，没有内置的支持，但是我知道有些异步框架应该能够完成相同的任务。 ，您可以达到相同的目标，但是在等待时允许其他异步功能运行。请注意，根据事件循环，您可能最终会等待更多或更少的时间。

我还清理了我提到的处理异常的问题；现在，它总是返回该函数的结果（如果有一个结果），并且它将重新引发该异常，而不会丢失任何回溯（如果有的话）。这还使用了仅Python 3的功能；我已经对如何在Python 2中进行操作发表了评论。

注意，我对asyncio并不熟悉，因为我从来没有在那里做过任何认真的开发工作，因此我可能没有这个一段代码完全正确；理论应该是正确的。

@retry(retry_count=5, delay=5)
def is_user_exist(username):
    try:
        pwd.getpwnam(username)
        log.info("User %s exist" % username)
        return True
    except KeyError:
        log.exception("User %s does not exist." % username)
        return False

#2 楼

我唯一注意到的是retry的行为可能会不一致。

让我解释一下：如果在func内部检查它成功，则会返回
while。另一方面，如果在None之外成功执行
，它将返回while返回的任何内容（在您的示例func中）。您不想拥有它。.


所以我建议稍微重新编码：

def retry(func, *func_args, **kwargs):
    retry_count = kwargs.get("retry_count", 5)
    delay = kwargs.get("delay", 5)
    while retry_count > 1:
        if func(*func_args):
            return
        log.debug("waiting for %s seconds before retyring again")
        sleep(delay)
        retry_count = retry_count - 1

    return func(*func_args)

如果愿意，可以替换上面的True循环，以得到一些幻想：只要不返回for就没有关系。

#3 楼

理论

您的retry函数与any的结构非常相似。

仅保留基本内容，您可以将retry编写为： br />
代码

如果要kwargs，log和sleep，可以编写：

any(func(*func_args) for _ in range(count))

注意事项



log.debug和time.sleep都是虚假的，所以当且仅当func or log or time是真实的时，func才是真实的。 。否则，它们将传递给dict.pop。

完整代码

def retry(func, *func_args, **kwargs):
    count = kwargs.pop("count", 5)
    delay = kwargs.pop("delay", 5)
    return any(func(*func_args, **kwargs)
               or log.debug("waiting for %s seconds before retyring again" % delay)
               or time.sleep(delay)
               for _ in range(count))

#4 楼

以下是您当前设置的一些问题：


被重试的函数不能使用关键字参数。在大多数情况下，可以很容易地解决此问题，但是允许函数接受类似delay的参数会更加复杂。这使得避免重复变得更加困难。
您最终可能会在日志中连续出现5次相同的回溯。当您必须添加额外的行时，这对于某些函数将是令人讨厌的，对于返回值有效的其他函数则是可怕的。我前一阵子写了这个装饰器，并愉快地在几个地方使用了它。这个想法与Dannnno的答案类似，但是我只在出现异常后重试，而不注意返回值。 br />

def retry(num_attempts=3, exception_class=Exception, log=None, sleeptime=1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for i in range(num_attempts):
                try:
                    return func(*args, **kwargs)
                except exception_class as e:
                    if i == num_attempts - 1:
                        raise
                    else:
                        if log:
                            log.error('Failed with error %r, trying again', e)
                        sleep(sleeptime)

        return wrapper

    return decorator

这是另一个示例，我仅在调用站点重试，而不更改函数的定义：
您的案件有点不寻常，因为其中涉及一个例外，但您最终不想提出该例外。您也许可以按照以下方式重新编写代码：

from requests.exceptions import ConnectionError

@retry(5, ConnectionError, log)
def request(self, method, url='', **kwargs):
    ...

成为：当条件为假但不涉及异常的情况下要重试时，可以显式引发异常：

在谈论如何编写一个非常通用且可广泛应用的功能，但是我们只有一个用例可以应用它。如果您有其他要重试的代码示例，则可以使此讨论更为有用。

#5 楼

我很惊讶没有人提到韧性库。

它完全可以满足您的要求+已有一个异步内置的实现。您还可以使用相当宽容的参数（等待然后重试，等待x时间等）。

#6 楼

您可以通过range（）通过简单的for循环替换while循环和retry_count倒数计时

def retry(func, *func_args, **kwargs):
    retry_count = kwargs.get("retry_count", 5)
    delay = kwargs.get("delay", 5)
    for _ in range(retry_count):
        if func(*func_args):
            return
        log.debug("waiting for %s seconds before retyring again")
        sleep(delay)

    return func(*func_args)