欢迎访问我的网站,希望内容对您有用,感兴趣的可以加入我们的社群。

Python实用模块(二十四)tenacity

实用模块 迷途小书童 4年前 (2020-11-04) 4843次浏览 0个评论

软硬件环境

  • windows 10 64bits
  • anaconda with python 3.7
  • tenacity

视频看这里

此处是 youtube 的播放链接,需要科学上网。喜欢我的视频,请记得订阅我的频道,打开旁边的小铃铛,点赞并分享,感谢您的支持。

简介

在实际应用中,经常会碰到在 web 请求时,因为网络的不稳定,会有请求超时的问题,这时候,一般都是自己去实现重试请求的逻辑,直到得到响应或者超时。虽然这样的逻辑并不复杂,但是代码写起来却不那么优雅,不那么 pythonic

tenacity 是一个重试库,使用 python 语言编写,它能够让我们在任务的重试操作中变得非常简单,使用的是 Apache 2.0 开源协议。

tenacity 有如下特性

  • 装饰器API
  • 可指定停止条件
  • 可指定等待条件
  • 自定义异常时的重试
  • 自定义特定返回值的重试
  • 在协程中使用

模块安装

使用 pip 安装 tenacity

pip install tenacity

示例代码

无条件重试

这是 tenacity 最基本的用法,在 task 方法中使用装饰器 @retry,当 task 出现异常时,我们就重新运行 task,这里没加任何限制,如果异常一直出现,task 就会一直运行下去

from tenacity import retry

@retry
def task():
    print("task running ... ")
    raise Exception

task()

执行上述代码,得到

task running ... 
task running ... 
task running ... 
task running ... 
task running ... 
task running ... 
task running ... 
task running ... 
task running ... 
task running ... 
task running ... 
task running ... 
task running ...
.
.
.

设定停止条件

通过方法 stop_after_attempt 指定重试的次数,如下的3次

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
def task():
    print("task running ... ")
    raise Exception

task()

或者使用方法 stop_after_delay 指定重试多长时候后停止,如下的3秒

from tenacity import retry, stop_after_delay

@retry(stop=stop_after_delay(3))
def task():
    print("task running ... ")
    raise Exception

task()

还可以将 stop_after_delaystop_after_attempt 组合起来用,如下的代码,只要其中一个条件满足,task 就停止运行

from tenacity import retry, stop_after_attempt, stop_after_delay

@retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
def task():
    print("task running ... ")
    raise Exception

task()

设定等待时间

使用方法 wait_fixed 来指定重试时等待的时间,如下代码中的3秒,每一次重试 task 前都要等待3秒钟

from tenacity import retry, wait_fixed

@retry(wait=wait_fixed(3))
def task():
    print("task running ... ")
    raise Exception

task()

使用方法 wait_random(min, max),在 minmax 之间随机取值,每一次 task 重试前就等待这个随机值,单位是秒

from tenacity import retry, wait_random

@retry(wait=wait_random(min=1, max=3))
def task():
    print("task running ... ")
    raise Exception

task()

当然,上面2中种方法也是可以结合起来用的

from tenacity import retry, wait_fixed, wait_random

@retry(wait=wait_fixed(3) + wait_random(0, 2))
def task():
    print("task running ... ")
    raise Exception

task()

什么情况下重试

可以通过 retry_if_exception_type 指定特定类型的异常出现时,任务才重试

from tenacity import retry, retry_if_exception_type

@retry(retry=retry_if_exception_type(IOError))
def task():
    print("task running ... ")
    raise Exception

task()

重试错误后的异常抛出

出现异常后,会进行重试,若重试后还是失败,默认情况下,往上抛出的异常会变成 RetryError,而不是最根本的原因。因此可以加一个参数 reraise=True,使得当重试失败后,往外抛出的异常还是原来的那个异常。

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3), reraise=True)
def task():
    print("task running ... ")
    raise Exception

task()

tenacity

在重试前执行动作

tenacity 可以在任务重试前后执行某些动作,这里以加日志为例

from tenacity import retry, stop_after_attempt, before_log
import logging
import sys

logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)

@retry(stop=stop_after_attempt(3), before=before_log(logger=logger, log_level=logging.DEBUG))
def task():
    print("task running ... ")
    raise Exception

task()

tenacity

重试后的操作类似,如下

from tenacity import retry, stop_after_attempt, after_log
import logging
import sys

logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
logger = logging.getLogger(__name__)

@retry(stop=stop_after_attempt(3), after=after_log(logger=logger, log_level=logging.DEBUG))
def task():
    print("task running ... ")
    raise Exception

task()

tenacity

Python实用模块专题

更多有用的 python 模块,请移步

https://xugaoxiang.com/category/python/modules/

参考资料

喜欢 (1)

您必须 登录 才能发表评论!