python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

前言

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

2、在与接口的通信过程中,为了防止由于网络不稳定情况,造成请求错误或者超时等问题,或者其他不可控因素等造成功能性问题,代码中一般都会加入重试功能以增加代码的健壮性。

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

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

5、 tenacity 库的特性:

  • 通用装饰器API
    [En]

    * Universal decorator API

  • 指定停止条件(即限制尝试次数)
    [En]

    * specify stop condition (that is, limit the number of attempts)

  • 指定等待条件(即两次尝试之间的指数退避睡眠)
    [En]

    * specify waiting conditions (that is, exponential Backoff sleep between attempts)

  • 自定义重试异常
    [En]

    * Custom retry exception

  • 自定义预期结果重试
    [En]

    * customize the retry of the expected result

  • 重试协作过程
    [En]

    * retry the collaboration process

  • 使用上下文管理器重试代码块
    [En]

    * retry the code block using the context manager

6、在编写程序尤其是与网络请求相关的程序,如调用web接口、运行网络爬虫等任务时,经常会遇到一些偶然发生的请求失败的状况,这种时候如果我们仅仅简单的捕捉错误然后跳过对应任务,肯定是不严谨的,尤其是在网络爬虫中,会存在损失有价值数据的风险。

安装

pip install tenacity

使用

1、基本重试(无条件重试,重试之间无间隔,报错之后就立马重试)

tenacity 库的错误重试核心功能由其 retry 装饰器来实现,默认在不给 retry 装饰器传参数时,它会在其所装饰的函数运行过程抛出错误时不停地重试下去。

from tenacity import retry

@retry
def never_give_up_never_surrender():
    print("无条件重试,重试之间无间隔,报错之后就立马重试")
    raise Exception

if __name__ == '__main__':
    never_give_up_never_surrender()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

2、停止重试

设置重试次数:

有时,我们对重试功能逻辑错误的容忍度是有限的。例如,当我们调用网络接口时,如果我们连续n次失败,我们可能会认为任务本身存在缺陷。不可能通过重试使其恢复正常。

[En]

Sometimes there is a limit to our tolerance for a retry of a function logic error. For example, when we call a network interface, if we fail n times in a row, we may think that the task itself is flawed. It is not possible to make it normal by retrying.

这种时候我们可以利用 tenacity 库中的 stop_after_attempt 函数,作为 retry() 中的 stop 参数传入,从而为我们”无尽”的错误重试过程添加一个终点,其中 stop_after_attempt() 接受一个整数输入作为「最大重试」的次数:

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(max_attempt_number=7))
def stop_after_7_attempts():
    print("重试7次后停止")
    raise Exception

if __name__ == '__main__':
    stop_after_7_attempts()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

设置时间限制:

tenacity 库还为我们提供了 stop_after_delay() 函数来设置整个重试过程的最大耗时,超出这个时长也会结束重试过程:

from tenacity import retry, stop_after_delay

@retry(stop=stop_after_delay(10))
def stop_after_10_s():
    print("10秒后停止重试")
    raise Exception

if __name__ == '__main__':
    stop_after_10_s()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

组合多个停止条件:

如果我们的任务需要添加最大重试次数最大超时限制,我们只需要使用|操作符将韧度库中的不同限制组合在一起,并传入retry()的Stop参数。

[En]

If our task needs to add * maximum number of retries * and * maximum timeout limit * , we only need to use the | operator to combine different restrictions in the tenacity library and pass in the stop parameter of retry ().

将 stop_after_delay 函数和 stop_after_attempt 函数组合起来用,如下的代码,只要其中一个条件满足,函数就抛出异常。

例如,当我们的函数执行重试超过3秒或超过5次时,我们可以结束重试:

[En]

For example, when our function executes a retry for more than 3 seconds or more than 5 times, we can end the retry:

from tenacity import retry, stop_after_delay, stop_after_attempt

@retry(stop=(stop_after_delay(max_delay=3) | stop_after_attempt(max_attempt_number=5)))
def stop_after_3_s_or_5_retries():
    print("10秒后或者5次重试后停止重试")
    raise Exception

if __name__ == '__main__':
    stop_after_3_s_or_5_retries()

运行结果:

[En]

Running result:

如您所见,在下面的结果中,达到了最大重试5次的限制,从而结束了函数执行的重试过程。

[En]

As you can see, in the following results, the limit of “maximum retry 5 times” is reached, thus ending the retry process of function execution.

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

3、重试前等待(设置相邻函数重试之间的时间间隔)

在某些情况下,我们不希望在每次重试引发错误后立即开始下一次重试。例如,为了在爬虫任务中更好地伪装我们的程序,坚韧性库提供了一系列非常有用的函数。通过使用retry()修饰符中的Wait参数,可以正确处理相邻重试之间的间隔。还有两种更实用的方法:

[En]

In some cases, we do not want to start the next retry immediately after each retry throws an error. For example, in order to better disguise our program in the crawler task, the tenacity library provides a series of very useful functions. With the wait parameter in the retry () decorator, the interval between adjacent retries can be handled properly. There are two more practical ways:

重试之前等待固定时间:

通过使用 tenacity 库中的 wait_fixed() 函数可以为相邻重试之间设置固定的等待间隔秒数,示例:

from tenacity import retry, wait_fixed

@retry(wait=wait_fixed(2))
def wait_2_s():
    print("每次重试前等待2秒")
    raise Exception

if __name__ == '__main__':
    wait_2_s()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

等待随机时间:

除了设置固定的时间间隔外, tenacity 库还可以通过 wait_random() 函数帮助我们为相邻重试设置均匀分布随机数,只需要设置好均匀分布的范围即可:

from tenacity import retry, wait_random

@retry(wait=wait_random(min=1, max=2))
def wait_random_1_to_2_s():
    print("每次重试前等待1-2秒之间")
    raise Exception

if __name__ == '__main__':
    wait_random_1_to_2_s()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

等待指数时间:

from tenacity import retry, wait_exponential

@retry(wait=wait_exponential(multiplier=1, min=4, max=10))
def wait_exponential_1():
    print("每次重试等待 2^x * multiplier 秒,x为重试次数,最小4秒,最多10秒")
    raise Exception

if __name__ == '__main__':
    wait_exponential_1()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

结合固定和抖动等待:

from tenacity import retry, wait_random, wait_fixed

@retry(wait=wait_fixed(3) + wait_random(0, 2))
def wait_fixed_jitter():
    print("每次重试前等待3到3+2秒之间")
    raise Exception

if __name__ == '__main__':
    wait_fixed_jitter()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

链式等待:

from tenacity import retry, wait_fixed, wait_chain

@retry(wait=wait_chain(*[wait_fixed(3) for i in range(3)] +
                        [wait_fixed(7) for i in range(2)] +
                        [wait_fixed(9)]))
def wait_fixed_chained():
    print("前3次等待3秒,接下来2次等待7秒,后面等待9秒")
    raise Exception

if __name__ == '__main__':
    wait_fixed_chained()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

4、自定义是否触发重试机制

tenacity 库中 retry() 装饰器的默认策略是当其所装饰的函数执行过程”抛出任何错误”时即进行重试,但有些情况下我们需要的可能是对特定错误类型的捕捉/忽略,亦或是对异常计算结果的捕捉。

捕捉或忽略特定的错误类型:

当函数中抛出的异常类型与 @retry() 装饰器中定义的异常类型相同时则执行重试策略,否则函数执行一次时直接抛出异常。(可以通过 retry_if_exception_type 函数指定的特定类型的异常出现时,任务才重试)

from tenacity import retry, retry_if_exception_type, retry_if_not_exception_type

@retry(retry=retry_if_exception_type(FileExistsError))
def demo_func7():
    raise TimeoutError

@retry(retry=retry_if_not_exception_type(FileNotFoundError))
def demo_func8():
    raise FileNotFoundError

if __name__ == '__main__':
    demo_func7()
    # demo_func8()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

自定义函数结果条件判断函数

通过可以编写额外的条件判断函数,配合 tenacity 库中的 retry_if_result() 装饰器,实现对函数的返回结果进行自定义条件判断,返回True时才会触发重试操作:

import random
from tenacity import retry, retry_if_result

@retry(retry=retry_if_result(lambda x: x >= 0.1))
def demo_func9():
    a = random.random()
    print(a)
    return a

# 记录开始时间
demo_func9()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

5、对函数的错误重试情况进行统计

被 tenacity 库中的 retry() 装饰的函数,可以通过打印其 retry.statistics 属性查看其历经的错误重试统计记录结果:

import random
from tenacity import retry, retry_if_result

@retry(retry=retry_if_result(lambda x: x >= 0.1))
def demo_func9():
    a = random.random()
    print(a)
    return a

# 记录开始时间
demo_func9()

print(demo_func9.retry.statistics)

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

6、重试错误后的异常抛出

异常发生后,函数将重试。如果重试仍然失败,则默认情况下,程序引发的最后一个异常将成为RetryError异常。如此一来,就会偏离节目的初衷。

[En]

After an exception occurs, the function will retry. If the retry still fails, by default, the last exception thrown by the program will become a RetryError exception. In this way, it will deviate from the original purpose of the program.

因此可以在 @retry() 装饰器中加一个参数 reraise=True ,使得当函数重试失败后,函数往外抛出的异常还是原来的那个异常。

1)不加 reraise=True 参数

from tenacity import retry, stop_after_attempt

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

task()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

2)加 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()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

7、在函数重试前执行动作

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

在函数执行前添加日志,然后执行函数,抛出异常后重新执行函数,然后在重新执行函数之前添加日志,依此类推。不,不,不。

[En]

Add the log before the function executes, then execute the function, then re-execute the function after throwing an exception, then add the log before re-executing the function, and so on. no, no, no.

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()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

8、在函数重试后执行操作

与函数重试之前的操作类似:

[En]

Similar to what you did before the function retried:

import time

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 ... ")
    time.sleep(2)
    raise Exception

task()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

9、自定义回调函数(当最后一次重试失败后,可以执行一个回调函数)

定义函数最后一次重试仍然失败调用的函数(只在最后一次重试失败时调用),回调函数应该接受一个被调用的参数 retry_state ,该参数包含有关当前重试调用的所有信息。

@retry 装饰器的定义回调函数的参数为: retry_error_callback=函数名 (回调函数一般返回最后一次函数重试的函数执行结果)

例1:

[En]

Example 1:

from tenacity import stop_after_attempt, retry, retry_if_result

def return_last_value(retry_state):
    """return the result of the last call attempt"""
    print('执行回调函数')
    print(retry_state.outcome.result())
    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=return_last_value,
       retry=retry_if_result(is_false))
def eventually_return_false():
    print('函数重试')
    return False

if __name__ == '__main__':
    eventually_return_false()

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

retry_state 参数是 RetryCallState 类的对象,具有以下属性:

  1. start_time(float) 重试开始时间戳
  2. retry_object(BaseRetrying) 重试对象
  3. fn(callable) 此重试调用包装的函数
  4. args(tuple) 此重试调用包装的函数参数
  5. kwargs(dict) 此重试调用包装的函数的关键字参数
  6. attempt_number(int) 当前尝试次数
  7. outcome(tenacity.FutureorNone) 函数产生的最后结果(结果或异常)
  8. outcome_timestamp(floatorNone) 最后结果的时间戳
  9. idle_for(float) 重试等待时间
  10. next_action(tenacity.RetryActionorNone) 由重试管理器决定的下一步操作

示例2:

[En]

Example 2:

from tenacity import retry, stop_after_attempt, retry_if_result

def return_last_value(retry_state):
    print("执行回调函数")
    return retry_state.outcome.result()  # 表示原函数的返回值

def is_false(value):
    print('执行《判断是否进行被装饰函数重试》的函数')
    return value is False

@retry(stop=stop_after_attempt(3),
       retry_error_callback=return_last_value,
       retry=retry_if_result(is_false))
def test_retry():
    print("等待重试.....")
    return False

print(test_retry())

运行结果:

[En]

Running result:

python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

https://blog.csdn.net/djstavaV/article/details/112261899

https://zhuanlan.zhihu.com/p/391812968

https://www.cnblogs.com/wuzhibinsuib/p/13443622.html

Original: https://www.cnblogs.com/hls-code/p/16490758.html
Author: 习久性成
Title: python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能

原创文章受到原创版权保护。转载请注明出处:https://www.johngo689.com/6608/

转载文章受原作者版权保护。转载请注明原作者出处!

(0)

大家都在看

免费咨询
免费咨询
扫码关注
扫码关注
联系站长

站长Johngo!

大数据和算法重度研究者!

持续产出大数据、算法、LeetCode干货,以及业界好资源!

2022012703491714

微信来撩,免费咨询:xiaozhu_tec

分享本页
返回顶部