Python的重试机制tenacity

1.安装

pip install tenacity -i https://pypi.douban.com/simple

最基本的重试

无条件重试，重试之间无间隔，报错之后就立马重试

from tenacity import retry

@retry
def test_retry():
    print("等待重试，重试无间隔执行")
    raise Exception

test_retry()

无条件重试，重试之前要等待2秒

from tenacity import retry, wait_fixed

@retry(wait=wait_fixed(2))
def test_retry():
    print("等待重试....")
    raise Exception

test_retry()

设置停止的基本条件

只重试7次，重试之间没，无间隔

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(7))
def test_retry():
    print("等待重试....")
    raise Exception

test_retry()

重试5秒之后不再重试

from tenacity import retry, stop_after_delay

@retry(stop=stop_after_delay(5))
def test_retry():
    print("等待重试....")
    raise Exception

test_retry()

重试7次，重试5秒，这个两个条件都需要满足

from tenacity import retry, stop_after_delay, stop_after_attempt

@retry(stop=(stop_after_delay(5) | stop_after_attempt(7)))
def test_retry():
    print("等待重试....")
    raise Exception

test_retry()

设置何时进行重试

在出现特定的错误/异常(比如请求超时）的情况下，在进行重试。

from requests import exceptions
from tenacity import retry, retry_if_exception_type

@retry(retry=retry_if_exception_type(exceptions.Timeout))
def test_retry():
    print("等待重试....")
    # raise Exception
    raise exceptions.Timeout
test_retry()

在满足自定义的条件时，再进行重试。

"""
如下实例中，当test_retry函数值为False时，在进行重试,并且同时满足重试三次
"""
from tenacity import retry, stop_after_attempt, retry_if_result

def is_false(value):
    return value is False

@retry(stop=stop_after_attempt(3), retry=retry_if_result(is_false))
def test_retry():
    print("等待重试.....")
    return False

test_retry()

重试后错误重新抛出

"""
当出现异常后，tenacity会进行重试，若重试失败之后，默认情况下，往上抛出的异常
会变成RetryError，而不是最开始的错误，因此可以加一个参数，reraise=True,使得，
重试失败后，往外抛出的异常还是原来的错误。
"""
from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(7), reraise=True)
def test_retry():
    print("等待重试.....")
    raise Exception

test_retry()

 设置回调函数

当最后一次重试失败之后，可以执行一个回调函数

from tenacity import retry, stop_after_attempt, retry_if_result

def return_last_value(retry_state):
    print("执行回调函数")
    return retry_state.outcome.result()  #表示原函数的返回值

def is_false(value):
    return value is False

@retry(stop=stop_after_attempt(3), retry_error_callback=return_last_value,
       retry=retry_if_result(is_false))
def test_retry():
    print("等待重试.....")
    return False

print(test_retry())