Python重试库 Tenacity详解(推荐)

这篇文章主要介绍了Python重试库Tenacity详解,文章围绕主题展开详细的内容介绍，具有一定的参考价值，需要的小伙伴可以参考一下

目录

• 1 Tenacity描述
• 2 如果发生异常就重试
• 3 设置停止重试的条件
• 设置重试的最大次数
• 还可以设置stop 时间
• 停止重试条件 进行组合
• 4 设置重试的间隔
• 5 重试触发的条件
• 针对具体的异常进行重试
• 针对函数的返回结果进行重试
• 6 定义重试失败回调函数
• 7 错误处理
• Basic usage
• Combination

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: Fail

The 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!

到此这篇关于Python重试库 Tenacity详解(推荐)的文章就介绍到这了,更多相关Python Tenacity 内容请搜索北冥有鱼以前的文章或继续浏览下面的相关文章希望大家以后多多支持北冥有鱼！