【摸鱼笔记】熟悉Tenacity，让运行更稳固

背景

在程序运行过程中，经常遇到需要处理报错的情况，大部分报错情况重试就能解决；

作为流行的编程语言，python提供了用于简化处理捕捉报错并重试的库，Tenacity 便是其中之一。

简介

Tenacity 是一个 Apache 2.0 许可的通用重试库，用 Python 编写，用于简化向几乎任何事物添加重试行为的任务。

它具有如下特性：

通用装饰器 API
指定停止条件（即尝试次数限制）
指定等待条件（即尝试之间的阻塞）
自定义重试异常
自定义对预期返回结果的重试
重试协程
使用上下文管理器重试代码块

安装

pip install tenacity

基本用法

Tenacity的基本用法是装饰器，该装饰器可以应用于函数或方法以实现重试操作；

示例

import tenacity@tenacity.retry(stop=tenacity.stop_after_attempt(3))
def func_execute():print("the long way rounds...")raise Exception("Exception!")try:func_execute()
except Exception as e:print(f"Exception: {e}")

运行结果

the long way rounds...
the long way rounds...
the long way rounds...
Exception: RetryError[<Future at 0x1874323ddc0 state=finished raised Exception>]

在func_execute函数中，手动抛出异常；

用@tenacity.retry装饰器来修饰函数func_execute；

如果@tenacity.retry中没有配置tenacity.stop_after_attempt将会无限循环；

示例中配置了重试策略，即在前三次尝试后停止重试tenacity.stop_after_attempt(3)；

由于配置了重试，Tenacity将在抛出异常时重试该函数，最多重试3次，在运行结果中可以看到函数中的 print 执行了三次。

配置选项

除了重试次数，Tenacity还提供了许多其他配置选项；

以下是一些常用的配置选项

retry

定义在哪些异常情况下执行重试，可以根据异常类型等自定义的条件；

tenacity.retry_if_exception_type(exception_types)

捕捉到与参数 exception_types 匹配的异常类型时重试

参数 exception_types ：装饰对象抛出匹配异常后才会重试，其他的会忽略

示例

import tenacity@tenacity.retry(retry=tenacity.retry_if_exception_type(IOError),stop=tenacity.stop_after_attempt(5))
def func_execute(param: str):print(param)raise IOError("IOError")try:func_execute("the long way rounds.")
except Exception as e:print(f"Exception: {e}")

运行结果

the long way rounds.
the long way rounds.
the long way rounds.
the long way rounds.
the long way rounds.
Exception: RetryError[<Future at 0x20d5d03f0e0 state=finished raised OSError>]

tenacity.retry_if_result(predicate)

装饰器装饰对象返回参数 predicate 匹配结果值为True时重试

参数 predicate ：用于判断返回值是否匹配的函数，其他的会忽略

示例

import tenacitydef retry_decide(result: str) -> bool:# 此处 func_execute 返回值为 None，# 返回给retry_if_result的值将是True# 不管是否报错都会重试return result is None@tenacity.retry(retry=tenacity.retry_if_result(retry_decide),stop=tenacity.stop_after_attempt(5))
def func_execute():print("执行 func_execute()")# 此处返回 Nonereturn Nonetry:result = func_execute()
except Exception as e:print(f"Exception: {e}")

运行结果

the long way rounds.
the long way rounds.
the long way rounds.
the long way rounds.
the long way rounds.
Exception: RetryError[<Future at 0x20d5d03f0e0 state=finished raised OSError>]

stop

定义停止的条件，可以根据重试次数、重试时间等条件；

tenacity.stop_after_attempt(max_attempt_number:int)

参数 max_attempt_number ：最大尝试次数，即超过尝试次数后抛出异常

示例

import tenacity@tenacity.retry(stop=tenacity.stop_after_attempt(5))
def func_execute():print("执行 func_execute()")raise Exception("抛出")try:result = func_execute()
except Exception as e:print(f"Exception: {e}")

执行 func_execute()
执行 func_execute()
执行 func_execute()
执行 func_execute()
执行 func_execute()
Exception: RetryError[<Future at 0x18ccf4fea80 state=finished raised Exception>]

tenacity.stop_after_delay(max_delay:int)

参数 max_delay ：最大尝试时长（单位为秒），即超过尝试时长后抛出异常

示例

import datetime as dtimport tenacity@tenacity.retry(stop=tenacity.stop_after_delay(5),wait=tenacity.wait_fixed(2))
def func_execute():print("执行 func_execute()", dt.datetime.now().strftime("%H:%M:%S"))raise Exception("抛出")try:result = func_execute()
except Exception as e:print(f"Exception: {e}")

运行结果

执行 func_execute() 16:40:02
执行 func_execute() 16:40:04
执行 func_execute() 16:40:06
执行 func_execute() 16:40:08
Exception: RetryError[<Future at 0x1d6622ae810 state=finished raised Exception>]

wait

定义每次重试之间的等待时间，一般是固定的等待时间，也可以设置每次递增；

tenacity.wait_fixed(wait)

参数 wait :每次重试的间隔时间

示例

import datetime as dtimport tenacity@tenacity.retry(stop=tenacity.stop_after_attempt(2),wait=tenacity.wait_fixed(2))
def func_execute():print("执行 func_execute()", dt.datetime.now().strftime("%H:%M:%S"))raise Exception("抛出")try:result = func_execute()
except Exception as e:print(f"Exception: {e}")

运行结果

执行 func_execute() 16:41:19
执行 func_execute() 16:41:21
Exception: RetryError[<Future at 0x1e84b9ff4a0 state=finished raised Exception>]

tenacity.wait_random(min,max)

参数 min :随机间隔时间范围下限
参数 max :随机间隔时间范围上限

示例

import datetime as dtimport tenacity@tenacity.retry(stop=tenacity.stop_after_delay(10),wait=tenacity.wait_random(2, 5))
def func_execute():print("执行 func_execute()", dt.datetime.now().strftime("%H:%M:%S"))raise Exception("抛出")try:result = func_execute()
except Exception as e:print(f"Exception: {e}")

运行结果

执行 func_execute() 16:42:14
执行 func_execute() 16:42:16
执行 func_execute() 16:42:19
执行 func_execute() 16:42:22
执行 func_execute() 16:42:25
Exception: RetryError[<Future at 0x1bd3737ed50 state=finished raised Exception>]

before_sleep

在每次重试之前执行的操作，可以用于执行清理或日志记录等任务。

before_sleep_log(logger, log_level)

参数 logger ：日志记录器对象

参数 log_level ：日志记录等级 debug, info 等

示例

import datetime as dt
import logging
import sysimport tenacitylogging.basicConfig(stream=sys.stderr, level=logging.DEBUG)logger = logging.getLogger(__name__)@tenacity.retry(retry=tenacity.retry_if_exception_type(IOError),stop=tenacity.stop_after_attempt(3),wait=tenacity.wait_fixed(2),before_sleep=tenacity.before_sleep_log(logger, logging.DEBUG))
def func_execute(param: str):print(param, dt.datetime.now().strftime("%H:%M:%S"))raise IOError("IOError")try:func_execute("the long way rounds.")
except Exception as e:print(f"Exception: {e}")

运行结果

DEBUG:__main__:Retrying __main__.func_execute in 2.0 seconds as it raised OSError: IOError.
the long way rounds. 16:43:24
the long way rounds. 16:43:26
DEBUG:__main__:Retrying __main__.func_execute in 2.0 seconds as it raised OSError: IOError.
the long way rounds. 16:43:28
Exception: RetryError[<Future at 0x2357a3be090 state=finished raised OSError>]

reraise

是否重新引发异常，如果设置为True，则在达到最大重试次数后会引发原始异常。

示例

import tenacity@tenacity.retry(retry=tenacity.retry_if_exception_type(IOError),stop=tenacity.stop_after_attempt(3),reraise=True)
def func_execute(param: str):print(param)raise IOError("IOError")try:func_execute("the long way rounds.")
except Exception as e:print(f"Exception: {e}")

运行结果

the long way rounds.
the long way rounds.
the long way rounds.
Exception: IOError

组合重试条件

自定义重试策略，以满足指定条件下重试或停止重试。

示例

import tenacity# 在同一个 配置项中组合配置条件
@tenacity.retry(stop=tenacity.stop_after_delay(10) | tenacity.stop_after_attempt(5),retry=tenacity.retry_if_exception_type(IOError))
def func_execute():print("Operation with Custom Stop...")raise IOError("Failed!")try:func_execute()
except Exception as e:print(f"Exception: {e}")