Python重试库 Tenacity详解(推荐)
作者：阿常呓语 发布时间：2021-07-07 00:20:09
1 Tenacity描述
今天 给大家介绍一个Python 重试库，Tenacity 这个库 是我 这些年 使用的一个非常好的库，几乎满足了我所有的重试需求，非常棒的一个库。
在我们日常开发中，我们经常会遇到 网络问题 或者其他问题，导致 API请求失败，或者超时等出现， 这个出现是偶发性，不确定性的，我们如何保证系统的功能的正确性呢？ 第一种 显示 告诉 用户请求失败。 第二种 我们可以通过重试 几次 看看会不会成功，如果 能成功 就可以了。
因此在真实的项目开发中重试的库，也显得 尤其重要， 在工作中 我也有尝试写过一些 重试的装饰器函数，总感觉 差点意思，要不就是 考虑不全面，要么就是 实现太简单 ，不能针对具体的异常进行重试，不能设置一些条件重试等。 以及重试 还是失败了怎么处理？ 等等一系列的问题。
直到后来发现这个 宝藏库 tenacity ，几乎解决了我所有的痛点。
一起来了解这个库的基本用法
2 如果发生异常就重试
这个就是 如果发生了异常就一直重试下去：
from tenacity import retry, stop_after_attempt
@retry()
def never_gonna_give_you_up():
print("Retry forever ignoring Exceptions, don't wait between retries")
raise Exception("Unknown Error")
3 设置停止重试的条件
设置重试的最大次数
这里设置最多重试5次 ，失败后 就不在进行重试了。
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(5))
def never_gonna_give_you_up():
print("Retry forever ignoring Exceptions, don't wait between retries")
raise Exception("Unknown Error")
if __name__ == '__main__':
never_gonna_give_you_up()
还可以设置stop 时间
stop = stop_after_delay
Stop when the time from the first attempt >= limit 从第一次失败 尝试 到 多久后 停止尝试。
默认 stop_after_delay 单位是秒
import time
from tenacity import retry, stop_after_delay
@retry(stop=stop_after_delay(10))
def stop_after_10_s():
print("10秒后停止重试")
time.sleep(5)
raise Exception("Unknown Error")
if __name__ == '__main__':
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("10秒后或者5次重试后停止重试")
raise Exception("Unknown Error")
if __name__ == '__main__':
stop_after_10_s_or_5_retries()
4 设置重试的间隔
有的时候 重试 太频繁了，也不太好，可能因为 网络问题，过一段时间才能恢复好，
wait_fixed(seconds=10) 等几秒后重试， 然后 可以组合 stop 在一起使用。
from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3),wait=wait_fixed(3))
def never_gonna_give_you_up():
print("Retry max attempt:3, wait_fixed:3")
raise Exception("Unknown Error")
if __name__ == '__main__':
never_gonna_give_you_up()
5 重试触发的条件
针对具体的异常进行重试
对具体的异常进行重试 ,而不是所有异常进行重试. 有的时候 我们在写 业务代码的时候，可能会定义一些 自定义的异常，我们有时候 触发特定的异常 进行重试，而不是说 发生了异常 就进行 重试。 tenacity 这个库 早就想好了这点。
我自定义了一个异常 NotExistResourceErr ， 触发这个异常我才进行重试，重试最多三次。
# -*- coding: utf-8 -*-
from typing import Any
from tenacity import retry, stop_after_delay, stop_after_attempt, retry_if_exception_type
class NotExistResourceErr(Exception):
"""
定义一个自定义的异常
"""
code = str(400)
detail = "Parameter Error"
def __init__(self, code: str = None, detail: Any = None):
self.code = code
self.detail = detail
class Source:
def __init__(self, id):
self.id = id
@retry(retry=retry_if_exception_type(NotExistResourceErr), stop=stop_after_attempt(3))
def might_exist_error(task_id: int = 0):
print('might_exist_error begin ...')
if task_id < 0:
raise NotExistResourceErr()
return Source(task_id)
if __name__ == '__main__':
might_exist_error(-10)
retry_if_exception_type 这里就告诉 tenacity 在发生什么类型，才进行重试操作。 设置 stop_after_attempt(3) 重试3 次就停止了。
运行一下结果如下： 可以看出 重试了三次，第三次后 抛出了异常
如果在里面 这样 抛出一个异常的话，此时 tenacity 不会进行重试操作，而是 程序直接报错。
@retry(retry=retry_if_exception_type(NotExistResourceErr), stop=stop_after_attempt(3))
def might_exist_error(task_id: int = 0):
print('might_exist_error begin ...')
if task_id < 0:
raise Exception("Unknown Error")
return Source(task_id)
如果这样的话， 不会进行重试操作 ，而是直接抛出了异常。
针对函数的返回结果进行重试
有的时候 可能因为 某些原因 正常情况下，应该可以获取到值，但是返回结果为None, 异常情况 在 函数内部进行捕获处理了，这个时候 也想进行 重试 怎么办呢？
retry_if_result 这里可以传入一个函数，这个函数接收 might_return_none 返回值， 然后 这个函数 返回 True 就会 触发重试操作。
from tenacity import retry, stop_after_attempt, retry_if_result
def is_none(value):
"""Return True if value is None"""
return value is None
@retry(retry=retry_if_result(is_none), stop=stop_after_attempt(3))
def might_return_none():
print("返回 None 则重试")
return None
if __name__ == '__main__':
might_return_none()
类似与此相反的 retry_if_not_result 当结果 不符合预期 则进行重试操作。
6 定义重试失败回调函数
对于重要的业务的话， 如果重试几次后， 发现仍然是失败的状态，此时 我们可以 设置 一个回调函数， 比如 在回调函数中 发一个邮件 通知 相关 的人员，手动处理异常，检查问题等。
retry_error_callback 这里可以传入一个回调 函数 ， 回函函数中有一个参数 retry_state 最后一次 函数的返回的状态。
在回调函数中 可以做你想做的任何处理，发邮件 或者其他的操作。
from tenacity import stop_after_attempt, retry, retry_if_result
def send_email(msg):
print(msg)
# 回调函数
def callback(retry_state):
"""return the result of the last call attempt"""
# print(f"call function ... {retry_state}, {type(retry_state)}")
send_email('eventually_return_false eventually failed')
return retry_state.outcome.result()
def is_false(value):
"""Return True if value is False"""
return value is False
# will return False after trying 3 times to get a different result
@retry(stop=stop_after_attempt(3),
retry_error_callback=callback,
retry=retry_if_result(is_false))
def eventually_return_false():
print("failed ...")
return False
if __name__ == '__main__':
eventually_return_false()
pass
7 错误处理
有的时候 我们可能重试后失败了，想获取原来 程序代码中抛出 的异常。 我们可以使用 reraise 参数来 抛出异常。是一个bool 变量，
设置为True 会抛出原来的异常， 如果设置为False 则不会抛出原来的异常 ,会抛出 tenacity.RetryError
from tenacity import retry, stop_after_delay, stop_after_attempt, retry_if_exception_type, retry_if_result
class NotExistResourceErr(Exception):
"""
定义一个自定义的异常
"""
code = str(400)
detail = "Parameter Error"
def __init__(self, code: str = None, detail: Any = None):
self.code = code or self.code
self.detail = detail or self.detail
@retry(reraise=True , stop=stop_after_attempt(3))
def raise_my_exception():
raise NotExistResourceErr(detail="Fail",code='0000')
try:
raise_my_exception()
except NotExistResourceErr as e :
print(f'handle exception detail:{e.detail} code:{e.code}')
设置为 reraise=False 整个程序 就会挂掉，如下面的报错：
Traceback (most recent call last):
File "/Users/frank/code/venv38/venv/lib/python3.8/site-packages/tenacity/__init__.py", line 407, in __call__
result = fn(*args, **kwargs)
File "/Users/frank/code/py_proj/study-fastapi/my_retry_demo/hello6.py", line 42, in raise_my_exception
raise NotExistResourceErr("Fail")
__main__.NotExistResourceErr: FailThe above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/Users/frank/code/py_proj/study-fastapi/my_retry_demo/hello6.py", line 45, in <module>
raise_my_exception()
File "/Users/frank/code/venv38/venv/lib/python3.8/site-packages/tenacity/__init__.py", line 324, in wrapped_f
return self(f, *args, **kw)
File "/Users/frank/code/venv38/venv/lib/python3.8/site-packages/tenacity/__init__.py", line 404, in __call__
do = self.iter(retry_state=retry_state)
File "/Users/frank/code/venv38/venv/lib/python3.8/site-packages/tenacity/__init__.py", line 361, in iter
raise retry_exc from fut.exception()
tenacity.RetryError: RetryError[<Future at 0x7f990dddb4c0 state=finished raised NotExistResourceErr>]
8 推荐看下 作者的对这个库介绍
做了一些翻译，感兴趣的同学 可以关注一下。Python never gives up: the tenacity library 原文
Python never gives up: the tenacity library
A couple of years ago, I wrote about the Python retrying library. This library was designed to retry the execution of a task when a failure occurred.
几年前，我写过关于Python重试库的文章。这个库被设计用来在一个任务发生故障时重试执行。
I started to spread usage of this library in various projects, such as Gnocchi, these last years. Unfortunately, it started to get very hard to contribute and send patches to the upstream retrying project. I spent several months trying to work with the original author. But after a while, I had to conclude that I would be unable to fix bugs and enhance it at the pace I would like to. Therefore, I had to take a difficult decision and decided to fork the library.
在过去的几年里，我开始在各种项目中推广使用这个库，比如Gnocchi。不幸的是，向上游的retrying项目贡献和发送补丁开始变得非常困难。我花了几个月的时间试图与原作者合作。但一段时间后，我不得不得出结论，我将无法以我想要的速度来修复错误和增强它。因此，我不得不做出一个艰难的决定，决定fork这个库。
Here comes tenacity
I picked a new name and rewrote parts of the API of retrying that were not working correctly or were too complicated. I also fixed bugs with the help of Joshua, and named this new library tenacity. It works in the same manner as retrying does, except that it is written in a more functional way and offers some nifty new features.
tenacity 就这样诞生了，
我选了一个新的名字，并重写了重试的API中不能正确工作或过于复杂的部分。我还在Joshua的帮助下修复了一些错误，并将这个新的库命名为tenacity。它的工作方式与重试相同，只是它是以一种更实用的方式编写的，并提供一些有趣的新功能。
nifty /ˈnɪftɪ/
niftier 比较级
niftiest 最高级
ADJ If you describe something as nifty, you think it is neat and pleasing or cleverly done. 整洁而惹人喜爱的; 完成得精巧的
• Bridgeport was a pretty nifty place.
布里奇波特曾是个相当整洁、惹人喜爱的地方。
• It was a nifty arrangement, a perfect partnership.
这既是个绝佳的安排，又是个完美的合作。
Basic usage
The basic usage is to use it as a decorator:
import tenacity
@tenacity.retry
def do_something_and_retry_on_any_exception():
pass
This will make the function do_something_and_retry_on_any_exception be called over and over again until it stops raising an exception. It would have been hard to design anything simpler. Obviously, this is a pretty rare case, as one usually wants to e.g. wait some time between retries. For that, tenacity offers a large panel of waiting methods:
这将使函数do_something_and_retry_on_any_exception被反复调用，直到它停止引发异常。很难再设计出更简单的东西了。显然，这是一个相当罕见的情况，因为人们通常希望在重试之间等待一段时间。为此，tenacity提供了大量的等待方法。
import tenacity
@tenacity.retry(wait=tenacity.wait_fixed(1))
def do_something_and_retry():
do_something()
Or a simple exponential back-off method can be used instead:
或者可以用一个简单的指数回退法来代替。
import tenacity
@tenacity.retry(wait=tenacity.wait_exponential())
def do_something_and_retry():
do_something()
Combination
组合
What is especially interesting with tenacity, is that you can easily combine several methods. For example, you can combine tenacity.wait.wait_random with tenacity.wait.wait_fixed to wait a number of seconds defined in an interval:
tenacity特别有趣的是，你可以很容易地结合几种方法。例如，你可以把 tenacity.wait.wait_random 和 tenacity.wait.wait_fixed 结合起来，等待在一个时间间隔内定义的若干秒。
import tenacity
@tenacity.retry(wait=tenacity.wait_fixed(10) + wait.wait_random(0, 3))
def do_something_and_retry():
do_something()
This will make the function being retried wait randomly between 10 and 13 seconds before trying again.
这将使被重试的函数在再次尝试之前随机地等待10到13秒。
tenacity offers more customization, such as retrying on some exceptions only. You can retry every second to execute the function only if the exception raised by do_something is an instance of IOError, e.g. a network communication error.
tenacity提供了更多的自定义功能，比如只在某些异常情况下重试。你可以在do_something引发的异常是IOError的实例时，例如网络通信错误，才每秒重试执行函数。
import tenacity
@tenacity.retry(wait=tenacity.wait_fixed(1),
retry=tenacity.retry_if_exception_type(IOError))
def do_something_and_retry():
do_something()
You can combine several condition easily by using the | or & binary operators. They are used to make the code retry if an IOError exception is raised, or if no result is returned. Also, a stop condition is added with the stop keyword arguments. It allows to specify a condition unrelated to the function result of exception to stop, such as a number of attemps or a delay.
你可以通过使用|或&二进制操作符轻松地组合几个条件。它们被用来在引发IOError异常或没有返回结果时使代码重试。此外，还可以用stop关键字参数添加一个停止条件。它允许指定一个与要停止的异常的函数结果无关的条件，例如尝试的次数或延迟。
import tenacity
@tenacity.retry(wait=tenacity.wait_fixed(1),
stop=tenacity.stop_after_delay(60),
retry=(tenacity.retry_if_exception_type(IOError) |
tenacity.retry_if_result(lambda result: result == None))
def do_something_and_retry():
do_something()
The functional approach of tenacity makes it easy and clean to combine a lot of condition for various use cases with simple binary operators.
tenacity的函数式方法使得它可以很容易和干净地用简单的二进制运算符为各种使用情况组合大量的条件。
Standalone usage
独立使用
tenacity can also be used without decorator by using the object Retrying, which implements its main behavior and using its call method. This allows to call any function with different retry conditions, or to retry any piece of code that do not use the decorator at all – like code from an external library.
tenacity也可以在没有装饰器的情况下使用，通过使用Retrying对象，实现其主要行为并使用其调用方法。这允许调用任何具有不同重试条件的函数，或重试任何完全不使用装饰器的代码–如来自外部库的代码。
import tenacity
r = tenacity.Retrying(
wait=tenacity.wait_fixed(1),
retry=tenacity.retry_if_exception_type(IOError))
r.call(do_something)
This also allows you to re-use that object without creating one new one each time, saving some memory!
这也允许你重复使用该对象，而不需要每次都创建一个新的对象，从而节省一些内存。
I hope you’ll like it and will find it of some use. Feel free to fork it, report bugs, or ask for new features on its GitHub!
我希望你会喜欢它，并会发现它有一些用处。欢迎在GitHub上分享它，报告错误，或要求提供新的功能。
If you want to learn more about retrying strategy and how to handle failure, there’s even more in Scaling Python. Check it out!
来源：https://blog.csdn.net/u010339879/article/details/126323247
0
投稿猜你喜欢
实现一个树形表格的时候有多种方法：比如把 ztree 的树形直接拼接成表格，或者用强大的 jqgrid 实现，今天介绍一个比较轻量级的实现：- 前言最小二乘 * east Square Method，做为分类回归算法的基础，有着悠久的历史（由马里·勒让德于1806年提出）。它通过最小化
- 前段时间看到letcode上的元音字母字符串反转的题目，今天来研究一下字符串反转的内容。主要有三种方法：1.切片法（最简洁的一种）#切片法d
- 这里介绍的是直接在package.json中直接配置使用：1.基础配置"build": { // 这里是el
- 本文实例为大家分享了python爬虫爬取淘宝商品的具体代码，供大家参考，具体内容如下1、需求目标 ： 进去淘宝页面，搜索耐克关键词，抓取 商
- 有一个比较有意思的传参方式：比如在 demo1.py 中指定 action='store_true'的时候：parser.a
- 昨时要导一些数据，从网上搜到的。字段多时insert 语句生成的不完整了，还没有找到原因..有个缺点……就是标识种子的列 也insert了c
- 问题描述像这样的图，我想把右边的colorbar设置成和主图一样高度方法0. plt.colorbar参数介绍https://matplot
- 本文实例讲述了PHP版微信小店接口开发方法。分享给大家供大家参考，具体如下：首先 大家可以去下一份小店开发的 API接口 因为 下面所有的
- 一、Xpath概述1、Xpath简介Xpath（XML Path Language） 是一门在 XML 文档中查找信息的语言，可以
- Python的3.0版本，常被称为Python 3000，或简称Py3k。相对于Python的早期版本，这是一个较大的升级。为了不带入过多的
- 前言Axios 是一个基于 promise 的 HTTP 库，可以用在浏览器和 node.js 中。axios 是目前最优秀的 HTTP 请
- 引言Requests是Python中一个常用的第三方库，用于向Web服务器发起HTTP请求并获取响应。该库的使用简单，功能强大，被广泛应用于
- 在我们的网站建设中，为网站打造一个契合网站主题的个性化标志则是必需的，这直接关系到能否成功地塑造网站的品牌。这从某些角度看仍在网站推广的范畴
- 在上一篇文章中，简单介绍了下闭包(closure)和原型链，现在继续来研究闭包的内部机制。对了，所有的东西都参考自这篇文章：Javascri
- set函数也是python内置函数的其中一个，属于比较基础的函数。其具体介绍和使用方法，下面进行介绍。set() 函数创建一个无序不重复元素
- 这个东西算是我被这个shuffle坑了的一个总结吧！首先我得告诉你一件事，那就是pytorch中的tensor，如果直接使用random.s
- argparse 是python自带的命令行参数解析包，可以用来方便地读取命令行参数。一、传入一个参数import argpars
- 自己写了玩的一个小脚本，百度图片下载import reimport osimport requestsimport hashlibdef d
- 最近在做一个项目，用双通道神经网络，每个通道输入不同数据训练，具有相同label。开始没想到如何实现，网上很多例子都是单通道，即便找到双通道
