本文主要是介绍python tenacity 重试库，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

Tenacity 重试库

tenacity 是一个很强大的重试库，前身是 retrying ，因为 retrying 停止维护了，所以有人弄出了这个库。

英文文档地址：https://tenacity.readthedocs.io/en/latest/

安装：

pip install tenacity

1. 基础使用

from tenacity import retry


@retry
def never_gonna_give_you_up():
    print("Retry forever ignoring Exceptions, don't wait between retries")
    raise Exception

当函数抛出异常时，@retry 会无限重试

2. 停止重试

规定次数后停止

from tenacity import stop_after_attempt


@retry(stop=stop_after_attempt(3))
def stop_after_3_attempts():
    print("Stopping after 3 attempts")
    raise Exception


stop_after_3_attempts()

加上限制条件，如果函数一直抛出异常，则只会重试3次

规定多久后停止

from tenacity import stop_after_attempt, stop_after_delay


@retry(stop=stop_after_delay(5))
def stop_after_5_s():
    print("Stopping after 5 seconds")
    raise Exception

stop_after_5_s()

会在 5 秒钟内一直重试，时间一到就会停止

多条件判断

from tenacity import stop_after_attempt, stop_after_delay


@retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
def stop_after_10_s_or_5_retries():
    print("Stopping after 10 seconds or 5 retries")
    raise Exception

stop_after_10_s_or_5_retries()

使用 | 条件，可以将多个条件一起使用。如果任意一个条件满足，就会停止重试。

3. 重试的时间间隔

固定时间间隔

from tenacity import stop_after_attempt, stop_after_delay, wait_fixed


@retry(wait=wait_fixed(2))
def wait_2_s():
    print("Wait 2 second between retries")
    raise Exception

wait_2_s()

每次重试之后，延迟 2 秒再进行下一次重试

随机时间间隔

from tenacity import stop_after_attempt, stop_after_delay, wait_fixed, wait_random


@retry(wait=wait_random(min=1, max=2))
def wait_random_1_to_2_s():
    print("Randomly wait 1 to 2 seconds between retries")
    raise Exception

wait_random_1_to_2_s()

指数增长的时间时间

from tenacity import retry, stop_after_attempt, stop_after_delay, wait_fixed, wait_random, wait_exponential


@retry(wait=wait_exponential(multiplier=1, min=4, max=10))
def wait_exponential_1():
    print("Wait 2^x * 1 second between each retry starting with 4 seconds, then up to 10 seconds, then 10 seconds afterwards")
    raise Exception

wait_exponential_1()

以指数增长的时间间隔，每次重试后，间隔时间以指数增长。

计算公式： 2^x * multiplier ，x 是 重试的次数-1 。计算后的值要和 min, max 对比，如果值在这个范围里，则使用计算出的值。如果值大于范围，则使用 max ，如果值小于这个范围，则使用 min

固定时间+随机时间

from tenacity import retry, stop_after_attempt, stop_after_delay, wait_fixed, wait_random, wait_exponential


@retry(wait=wait_fixed(3) + wait_random(0, 2))
def wait_fixed_jitter():
    print("Wait at least 3 seconds, and add up to 2 seconds of random delay")
    raise Exception

wait_fixed_jitter()

给不同的次数分配不同的等待时间

from tenacity import retry, stop_after_attempt, stop_after_delay, wait_fixed, wait_random, wait_exponential,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("Wait 3s for 3 attempts, 7s for the next 2 attempts and 9s for all attempts thereafter")
    raise Exception

wait_fixed_chained()

4. 是否重试

4.1 根据错误类型重试

from tenacity import retry, retry_if_exception_type, retry_if_not_exception_type


class ClientError(Exception):
    """Some type of client error."""

@retry(retry=retry_if_exception_type(IOError))
def might_io_error():
    print("如果错误类型是 IOError, 则重试；其他错误类型则不会进行重试")
    raise Exception

@retry(retry=retry_if_not_exception_type(ClientError))
def might_client_error():
    print("如果错误类型不是 ClientError, 则进行重试；如果是 ClientError，则不会重试")
    raise ClientError('')

might_client_error()
might_io_error()

4.2 根据返回值进行重试

from tenacity import retry, retry_if_result


def judge_result(x):
	"""Return True if value is None"""
	return x is None


@retry(retry=retry_if_result(judge_result))
def might_return_none():
	print("Retry with no wait if return value is None")
	return None


might_return_none()

上面的例子中，我们 might_return_none() 函数的返回值会传递给 judge_result() 函数。如果 judge_result() 返回了 True 则会进行重试。

所以这里的关键是要写一个辅助函数，判断当前函数的返回值。并将这个辅助函数传递给 retry_if_result

4.3 多条件组合

from tenacity import retry, retry_if_result, retry_if_exception_type


def is_none_p(value):
	"""Return True if value is None"""
	return value is None

@retry(retry=(retry_if_result(is_none_p) | retry_if_exception_type()))
def might_return_none():
	print("Retry forever ignoring Exceptions with no wait if return value is None")
	return IOError('IO Error')


might_return_none()

否则任意一个条件满足，都会重试。

5. 错误处理

5.1 重新抛出原始异常

from tenacity import retry, retry_if_result, retry_if_exception_type, stop_after_attempt


class MyException(Exception):
	pass

@retry(reraise=True, stop=stop_after_attempt(3))
def raise_my_exception():
	raise MyException("Fail")

try:
	raise_my_exception()
except Exception as e:
	print(e)

reraise 参数可以设置重新抛出原始的异常。即当最后一次重试依然有异常时，则重新抛出原始异常，而不是默认的 RetryError

5.2 before 、 after

import logging
import sys

from tenacity import retry, stop_after_attempt, before_log

logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)

logger = logging.getLogger(__name__)


@retry(stop=stop_after_attempt(3), before=before_log(logger, logging.DEBUG))
def raise_my_exception():
	raise Exception("Fail")


raise_my_exception()

before 可以在重试之前，做一些事情。这里用来记录 log。导入了一个 before_log 的函数。

同理，after:

import logging
import sys

from tenacity import retry, stop_after_attempt, after_log

logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)

logger = logging.getLogger(__name__)


@retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG))
def raise_my_exception():
	raise Exception("Fail")


raise_my_exception()

6. 改变参数

from tenacity import retry, stop_after_attempt, retry_if_result


@retry(stop=stop_after_attempt(3))
def raise_my_exception():
	print('retrying.')
	raise Exception("Fail")

try:
	raise_my_exception.retry_with(stop=stop_after_attempt(4))()  # 运行时可以改变参数
except Exception:
	pass

也可以生成一个对象，将想要重试的函数作为参数传入：

from tenacity import retry, stop_after_attempt, retry_if_result, Retrying


def never_good_enough(arg1):
	print(arg1)
	raise Exception('Error')


def try_never_good_enough(max_attempts=3):
	retryer = Retrying(stop=stop_after_attempt(max_attempts), reraise=True)  # retryer 对象
	retryer(never_good_enough, 'I really do try')  # 将要重试的函数和参数传递进去

try_never_good_enough()

7. 异步

可以支持 async 函数：

import asyncio

from tenacity import retry, stop_after_attempt


@retry(stop=stop_after_attempt(3), reraise=True)
async def my_async_function(seconds):
	await asyncio.sleep(seconds)
	print(f'sleep {seconds}')
	raise Exception('Async Error')


loop = asyncio.new_event_loop()
loop.run_until_complete(my_async_function(1))

这篇关于python tenacity 重试库的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！