tenacity发生异常/失败/错误时重试retry机制，Python

Posted 2022-09-06 zhangphil

tenacity发生异常/失败/错误时重试retry机制，Python

安装：

pip install tenacity 

示例：

@retry
def non_stop():
    print("永不停息")
    raise Exception

代码运行后，将用不停息的输出：

永不停息
永不停息
永不停息

...

发生异常重试若干次后退出。

from tenacity import retry, stop_after_attempt


@retry(stop=stop_after_attempt(3))
def non_stop():
    print("永不停息")
    raise Exception

输出：

永不停息
永不停息
永不停息

上面代码发生异常重试3次后结束重试。

重试若干时间（秒）后退出：

import datetime

from tenacity import retry, stop_after_delay


@retry(stop=stop_after_delay(2))
def non_stop():
    t = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(t)
    raise Exception

上面代码发生异常错误重试2秒后结束重试。

发生异常重试若干次或者重试若干秒后结束：

import datetime

from tenacity import retry, stop_after_delay, stop_after_attempt


@retry(stop=(stop_after_delay(2) | stop_after_attempt(3)))
def retry_stop():
    t = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(t)
    raise Exception

上面代码重试3次或者重试2秒后结束。

发生异常后，等待一定时间后重试：

import datetime

from tenacity import retry, wait_fixed


@retry(wait=wait_fixed(3))
def retry_stop():
    t = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(t)
    raise Exception

以上代码发生异常后，等待3秒重试。

发生异常，等待一定随机时间后重试：

import datetime

from tenacity import retry, wait_random


@retry(wait=wait_random(min=2, max=5))
def exc_retry():
    t = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(t)
    raise Exception

以上代码，发生异常后，等待最少2秒，最多5秒的随机时间后重试。

发生异常后，等待一定量时间，再加上一定随机时间后重试：

import datetime

from tenacity import retry, wait_random, wait_fixed


@retry(wait=wait_fixed(2) + wait_random(1, 3))
def exc_retry():
    t = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(t)
    raise Exception

上面代码发生异常后，等待2秒的基础上再加上1到3秒之间随机时间总和后重试。

只有当发生特定异常后才重试：

import datetime

from tenacity import retry, retry_if_exception_type


@retry(retry=retry_if_exception_type(IOError))
def exc_retry():
    t = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(t)
    raise IOError

上面代码，只有发生IOError后才重试。

只有当满足条件为True时候，才重试：

import datetime

from tenacity import retry, retry_if_result


def is_value(value):
    print(value)
    return True


@retry(retry=retry_if_result(is_value))
def program_retry():
    t = datetime.datetime.now().strftime('%M:%S')
    print(t)
    return -1
    # raise Exception

program_retry()返回的值进入is_value()函数处理，只有当is_value()函数返回值为True时候才重试，如果is_value()返回为False，则停止重试。

查看异常信息统计：

import datetime

from tenacity import retry, stop_after_attempt


@retry(stop=stop_after_attempt(3))
def program_retry():
    t = datetime.datetime.now().strftime('%M:%S')
    print(t)
    raise Exception()


if __name__ == '__main__':
    try:
        program_retry()
    except:
        pass
    print(program_retry.retry.statistics)

输出：

23:32
23:32
23:32
'start_time': 你的程序启动时间, 'attempt_number': 3, 'idle_for': 0, 'delay_since_first_attempt': 0.0