1.官网说法:
Tenacity is an Apache 2.0 licensed general-purpose retrying library, written in Python, to simplify the task of adding retry behavior to just about anything. It originates from a fork of retrying which is sadly no longer maintained. Tenacity isn’t api compatible with retrying but adds significant new functionality and fixes a number of longstanding bugs.
Tenacity是一个Apache 2.0许可的通用重试库,用Python编写,简化了向任何事情添加重试行为的任务。
它起源于一个令人遗憾的不再维持的retrying。
Tenacity与retrying不兼容,但增加了重要的新功能并修复了许多长期存在的错误。
2.安装
pip install tenacity
3.基本重试,是对函数异常进行重试,知道成功为止
如:无条件,永久重试型
@retry
def never_give_up_never_surrender():
print("Retry forever ignoring Exceptions, don't wait between retries")
raise Exception
设置停止基本条件
@retry(stop=stop_after_attempt(7)) # 表示重试7次后停止
def stop_after_7_attempts():
print("Stopping after 7 attempts")
raise Exception
@retry(stop=stop_after_delay(10)) # 表示开始10秒后停止
def stop_after_10_s():
print("Stopping after 10 seconds")
raise Exception
@retry(stop=(stop_after_delay(10) | stop_after_attempt(5))) # 用| 进行多条件组合,满足任意一个就停止
def stop_after_10_s_or_5_retries():
print("Stopping after 10 seconds or 5 retries")
raise Exception
在重试之前等待
@retry(wait=wait_fixed(2)) # 表示在重试之间等待两秒
def wait_2_s():
print("Wait 2 second between retries")
raise Exception
其中wait_random(min=1, max=2)表示随机等待1秒或者两秒
wait=wait_exponential(multiplier=1, min=4, max=10)表示每次重试之间等待2 ^ x * 1秒,从4秒开始,然后最多10秒,然后是10秒
wait_fixed(3) + wait_random(0, 2)表示等待至少3秒钟,并添加最多2秒的随机延迟
wait_random_exponential(multiplier=1, max=60)表示每次重试之间随机等待最多2 ^ x * 1秒,直到范围达到60秒,然后随机最多60秒(当多个进程争用共享资源时,指数级增加的抖动有助于最大限度地减少冲突。)
wait_chain(*[wait_fixed(3) for i in range(3)] +
[wait_fixed(7) for i in range(2)] +
[wait_fixed(9)]))表示等待3次尝试3次,接下来的2次尝试等待7次,之后所有尝试等待9次
是否重试
我们有一些选项来处理引发特定或一般异常的重试,如此处的情况。
@retry(retry=retry_if_exception_type(IOError))
def might_io_error():
print("Retry forever with no wait if an IOError occurs, raise any other errors")
raise Exception
自定义重试
# 自定义回调
# 回调应该接受一个名为的参数retry_state,该参数包含有关当前重试调用的所有信息。
# 可以在所有重试失败后调用自定义回调函数,而不会引发异常(或者您可以重新提升或执行任何操作)
def return_last_value(retry_state):
"""return the result of the last call attempt"""
return retry_state.outcome.result() # 表示返回原函数的返回值
def is_false(value):
"""Return True if value is False"""
return value[0] in (False,)
# retry_if_result(fun)和retry_if_not_result(fun)分别定义了重试的条件(对比条件是被装饰函数的返回值
# ),其中的参数是条件的函数,如is_false()函数,不同之处是一个表示当条件满足则重试,一个是当条件不满足则重试
@retry(stop=stop_after_attempt(3),
retry_error_callback=return_last_value,
retry=retry_if_result(is_false)) # 表示重试3次,当被装饰函数的返回值满足is_false的条件时才重试,返回值是return_last_vcalue函数中定义的返回值
def eventually_return_false():
print(1)
return False,'a'
错误处理
@retry(reraise=True, stop=stop_after_attempt(3))
def raise_my_exception():
raise MyException("Fail") # 自定义错误
try:
raise_my_exception()
except MyException:
# timed out retrying
pass