持续创作，加速成长！这是我参与「掘金日新计划 · 6 月更文挑战」的第7天，点击查看活动详情
介绍
本文介绍另一个Python中的重试模块tenacity， 前面的文章中有介绍过Python的一个第三方包retry 包，也是一个重试模块，有兴趣的XDM可以移步链接进行查看。
和retry 一样，该模块同样可以实现在出现某些错误的情况下自动的重复执行代码，提高了我们开发程序的效率，使用该模块的方式通常情况下也是通过装饰器的方式进行使用。
安装和使用
安装
pip install tenacity
复制代码使用
基础重试
是该模块的基本用法，只要被装饰对象出现异常就重试，没有任何限制，只要出现异常，被装饰对象就会一直重复运行下去，直至运行返回正确结果。
from tenacity import retry
@retry
def never_gonna_give_you_up():
print("Retry forever ignoring Exceptions, don't wait between retries")
raise Exception
复制代码有停止条件的重试
代码不应该无条件的无限重试下去，可以通过该模块提供的一些方法规定何时终止重试
- 指定重试次数
通过指定重试次数规定何时停止重试，需要用到该模块下的stop _after_attempt。
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(7))
def stop_after_7_attempts():
print("Stopping after 7 attempts")
raise Exception
复制代码- 指定重试多长时间后停止重试
通过指定重试多长时间后停止重试，需要用到该模块下的stop_after_delay。
from tenacity import retry, stop_after_delay
@retry(stop=stop_after_delay(10))
def stop_after_10_s():
print("Stopping after 10 seconds")
raise Exception
stop_after_10_s()
复制代码- 将重试停止条件进行组合使用
可以将上述两个终止重试的条件进行组合使用，通过|对不同的条件进行分隔，只要满足任意一个条件，重试就会终止。
from tenacity import retry, stop_after_delay, stop_after_attempt
@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
复制代码重试时间间隔
对于代码来说，没有间隔的重试其实是不推荐的，我们可以在每次重试之间设定时间间隔。
- 指定重试时间间隔
可以使用wait_fixed 方法指定重试间隔。下述代码就是每次重试之间间隔2s。
from tenacity import retry, wait_fixed
@retry(wait=wait_fixed(2))
def wait_2_s():
print("Wait 2 second between retries")
raise Exception
wait_2_s()
复制代码- 重试时间间隔随机
我们可以让时间间隔变成随机的数值。
from tenacity import retry, wait_random
@retry(wait=wait_random(min=1, max=2))
def wait_random_1_to_2_s():
print("Randomly wait 1 to 2 seconds between retries")
raise Exception
复制代码- 多种时间间隔组合使用
上述时间间隔的方式是可以组合在一起使用的。下述代码示例表示每次重试之间的时间间隔至少是3s, 并且会在这3s之上随机增加0-2s。
from tenacity import retry, wait_random, wait_fixed
@retry(wait=wait_fixed(3) + wait_random(0, 2))
def wait_fixed_jitter():
print("Wait at least 3 seconds, and add up to 2 seconds of random delay")
raise Exception
复制代码何时重试
我们可以通过retry_if_exception_type来指定当特定类型的异常出现时，才会重试。并且异常类型我们是可以自定义的，比如下述代码的ClientError。
from tenacity import retry, retry_if_not_exception_type
class ClientError(Exception):
"""Some type of client error."""
@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(retry=retry_if_not_exception_type(ClientError))
def might_client_error():
print("Retry forever with no wait if any error other than ClientError occurs. Immediately raise ClientError.")
raise Exception
复制代码另外我们还可以通过函数的返回结果来控制重试。
from tenacity import retry, retry_if_result
def is_none_p(value):
"""Return True if value is None"""
return value is None
@retry(retry=retry_if_result(is_none_p))
def might_return_none():
print("Retry with no wait if return value is None")
复制代码当然上述两种条件可以结合使用。
from tenacity import retry, retry_if_result, retry_if_exception_type
def is_none_p(value):
"""Return True if value is None"""
return value is None
@retry(retry=(retry_if_result(is_none_p) | retry_if_exception_type()))
def might_return_none():
print("Retry forever ignoring Exceptions with no wait if return value is None")
复制代码在重试之前执行某些任务
我们可以通过before回调函数在重试之前执行某些任务，这里就以记录日志为例。
from tenacity import retry, before_log
import logging
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), before=before_log(logger, logging.DEBUG))
def raise_my_exception():
raise MyException("Fail")
复制代码同样的也可以通过after回调函数在函数重试之后执行某些任务，同样以记录日志为例，即在重试之后记录。
from tenacity import retry, before_log, after_log
import logging
logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG))
def raise_my_exception():
raise MyException("Fail")
复制代码统计数据
可以使用重试的某些方法对函数的重试进行统计得到重试过程中的一些记录或者数据
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
def raise_my_exception():
raise ValueError("Fail")
try:
raise_my_exception()
except Exception:
pass
print(raise_my_exception.retry.statistics) # {'start_time': 157353.265, 'attempt_number': 3, 'idle_for': 0, 'delay_since_first_attempt': 0.0}
复制代码总结
重试的模块有很多，大家根据实际情况选择合适的好用的即可。