Tenacity

.. image:: https://img.shields.io/pypi/v/tenacity.svg :target: https://pypi.python.org/pypi/tenacity

.. image:: https://circleci.com/gh/jd/tenacity.svg?style=svg :target: https://circleci.com/gh/jd/tenacity

.. image:: https://img.shields.io/endpoint.svg?url=https://api.mergify.com/badges/jd/tenacity&style=flat :target: https://mergify.io :alt: Mergify Status

请参阅 tenacity文档 <https://tenacity.readthedocs.io/en/latest/>_ 以获得更好的体验。

Tenacity是一个基于Apache 2.0许可的通用重试库,用Python编写,旨在简化为几乎任何事物添加重试行为的任务。它源自retrying的一个分支<https://github.com/rholder/retrying/issues/65>,遗憾的是该项目不再维护<https://julien.danjou.info/python-tenacity/>。Tenacity与retrying的API不兼容,但增加了大量新功能并修复了许多长期存在的bug。

最简单的用例是在出现Exception时重试一个不稳定的函数,直到返回一个值。

.. testcode::

import random
from tenacity import retry

@retry
def do_something_unreliable():
    if random.randint(0, 10) > 1:
        raise IOError("糟糕,一切都崩溃了!!!111one")
    else:
        return "太棒了!"

print(do_something_unreliable())

.. testoutput:: :hide:

太棒了!

.. toctree:: :hidden: :maxdepth: 2

changelog
api

特性

• 通用装饰器API
• 指定停止条件(例如限制尝试次数)
• 指定等待条件(例如尝试之间的指数退避睡眠)
• 自定义对异常的重试
• 自定义对预期返回结果的重试
• 对协程进行重试
• 使用上下文管理器重试代码块

安装

安装 tenacity 非常简单:

.. code-block:: bash

$ pip install tenacity

示例

基本重试

.. testsetup::

    import logging
    #
    # 注意:以下导入仅用于演示方便。
    # 生产代码应始终显式导入所需的名称。
    #
    from tenacity import *

    class MyException(Exception):
        pass

如上所示,默认行为是在出现异常时永远重试,不等待。

.. testcode::

    @retry
    def never_gonna_give_you_up():
        print("忽略异常并永远重试,重试之间不等待")
        raise Exception

停止
~~~~~~~~

让我们少一些坚持,设置一些界限,比如在放弃之前的尝试次数。

.. testcode::

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

我们没有整天的时间,所以让我们为重试设置一个时间界限。

.. testcode::

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

如果你有严格的截止时间,超过延迟时间是不可接受的,那么你可以在即将超过延迟时间前一次尝试就放弃重试。

.. testcode::

    @retry(stop=stop_before_delay(10))
    def stop_before_10_s():
        print("在10秒前1次尝试停止")
        raise Exception

你可以使用`|`运算符组合多个停止条件:

.. testcode::

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

重试前等待

大多数事物不喜欢被尽可能快地轮询,所以让我们在重试之间等待2秒。

.. testcode::

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

有些事情在注入一些随机性时表现最佳。

.. testcode::

@retry(wait=wait_random(min=1, max=2))
def wait_random_1_to_2_s():
    print("重试之间随机等待1到2秒")
    raise Exception

再说一次,在重试分布式服务和其他远程端点时,指数退避难以超越。

.. testcode::

@retry(wait=wait_exponential(multiplier=1, min=4, max=10))
def wait_exponential_1():
    print("每次重试之间等待2^x * 1秒,从4秒开始,然后最多10秒,之后保持10秒")
    raise Exception

再说一次,在重试分布式服务和其他远程端点时,结合固定等待和抖动(以帮助避免惊群效应)也难以超越。

.. testcode::

@retry(wait=wait_fixed(3) + wait_random(0, 2))
def wait_fixed_jitter():
    print("至少等待3秒,并添加最多2秒的随机延迟")
    raise Exception

当多个进程争用共享资源时,指数增加的抖动有助于最小化冲突。

.. testcode::

@retry(wait=wait_random_exponential(multiplier=1, max=60))
def wait_exponential_jitter():
    print("每次重试之间随机等待最多2^x * 1秒,直到范围达到60秒,之后随机等待最多60秒")
    raise Exception

有时需要构建一个退避链。

.. testcode:: @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

是否重试

我们有几个选项来处理引发特定或一般异常的重试情况，如下所示。

.. testcode::

    class ClientError(Exception):
        """某种客户端错误。"""

    @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 Exception

我们还可以使用函数的返回结果来改变重试行为。

.. testcode::

    def is_none_p(value):
        """如果值为None则返回True"""
        return value is None

    @retry(retry=retry_if_result(is_none_p))
    def might_return_none():
        print("如果返回值为None则无等待重试")

另请参阅以下方法：

.. testcode::

    retry_if_exception
    retry_if_exception_type
    retry_if_not_exception_type
    retry_unless_exception_type
    retry_if_result
    retry_if_not_result
    retry_if_exception_message
    retry_if_not_exception_message
    retry_any
    retry_all

我们还可以组合多个条件：

.. testcode::

    def is_none_p(value):
        """如果值为None则返回True"""
        return value is None

    @retry(retry=(retry_if_result(is_none_p) | retry_if_exception_type()))
    def might_return_none():
        print("如果返回值为None，则忽略异常并无等待地永久重试")

还支持任意组合stop、wait等参数，让您可以自由搭配。

您还可以通过抛出`TryAgain`异常来在任何时候显式重试：

.. testcode::

   @retry
   def do_something():
       result = something_else()
       if result == 23:
          raise TryAgain

错误处理

通常，当您的函数最后一次失败（并且根据您的设置不会再次重试）时，会抛出RetryError。您的代码遇到的异常将显示在堆栈跟踪的中间某处。

如果您希望在堆栈跟踪的末尾（最容易看到的地方）看到您的代码遇到的异常，可以设置reraise=True。

.. testcode::

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

try:
    raise_my_exception()
except MyException:
    # 重试超时
    pass

重试前后的操作和日志记录

通过使用before回调函数，可以在调用函数的任何尝试之前执行操作：

.. testcode::

    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, logging.DEBUG))
    def raise_my_exception():
        raise MyException("失败")

同样，可以在失败的调用之后执行操作：

.. testcode::

    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, logging.DEBUG))
    def raise_my_exception():
        raise MyException("失败")

还可以只记录将要重试的失败。通常重试会在等待间隔之后发生，所以关键字参数称为"before_sleep"：

.. testcode::

    import logging
    import sys

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

    logger = logging.getLogger(__name__)

    @retry(stop=stop_after_attempt(3),
           before_sleep=before_sleep_log(logger, logging.DEBUG))
    def raise_my_exception():
        raise MyException("失败")

统计信息
~~~~~~~~

您可以通过使用附加到函数的`statistics`属性来访问有关函数重试的统计信息：

.. testcode::

    @retry(stop=stop_after_attempt(3))
    def raise_my_exception():
        raise MyException("失败")

    try:
        raise_my_exception()
    except Exception:
        pass

    print(raise_my_exception.statistics)

.. testoutput::
   :hide:

   ...

自定义回调
~~~~~~~~~~

您还可以定义自己的回调。回调应接受一个名为"retry_state"的参数，该参数包含有关当前重试调用的所有信息。

例如，您可以在所有重试失败后调用自定义回调函数，而不抛出异常（或者您可以重新抛出异常或做任何其他操作）

.. testcode::

    def return_last_value(retry_state):
        """返回最后一次调用尝试的结果"""
        return retry_state.outcome.result()

    def is_false(value):
        """如果值为False则返回True"""
        return value is False

    # 尝试3次获取不同结果后将返回False
    @retry(stop=stop_after_attempt(3),
           retry_error_callback=return_last_value,
           retry=retry_if_result(is_false))
    def eventually_return_false():
        return False

RetryCallState
~~~~~~~~~~~~~~

"retry_state"参数是:class:`~tenacity.RetryCallState`类的对象。

其他自定义回调
~~~~~~~~~~~~~~

还可以为其他关键字参数定义自定义回调。

.. function:: my_stop(retry_state)

   :param RetryCallState retry_state: 有关当前重试调用的信息
   :return: 是否应停止重试
   :rtype: bool

.. function:: my_wait(retry_state)
:param RetryCallState retry_state: 关于当前重试调用的信息
:return: 下次重试前需要等待的秒数
:rtype: float

.. function:: my_retry(retry_state)

   :param RetryCallState retry_state: 关于当前重试调用的信息
   :return: 是否应继续重试
   :rtype: bool

.. function:: my_before(retry_state)

   :param RetryCallState retry_state: 关于当前重试调用的信息

.. function:: my_after(retry_state)

   :param RetryCallState retry_state: 关于当前重试调用的信息

.. function:: my_before_sleep(retry_state)

   :param RetryCallState retry_state: 关于当前重试调用的信息

以下是一个使用自定义 ``before_sleep`` 函数的示例:

.. testcode::

    import logging

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

    logger = logging.getLogger(__name__)

    def my_before_sleep(retry_state):
        if retry_state.attempt_number < 1:
            loglevel = logging.INFO
        else:
            loglevel = logging.WARNING
        logger.log(
            loglevel, '正在重试 %s: 第 %s 次尝试以 %s 结束',
            retry_state.fn, retry_state.attempt_number, retry_state.outcome)

    @retry(stop=stop_after_attempt(3), before_sleep=my_before_sleep)
    def raise_my_exception():
        raise MyException("失败")

    try:
        raise_my_exception()
    except RetryError:
        pass


在运行时更改参数

通过使用包装函数附加的 retry_with 函数，你可以在调用时根据需要更改重试装饰器的参数:

.. testcode::

@retry(stop=stop_after_attempt(3))
def raise_my_exception():
    raise MyException("失败")

try:
    raise_my_exception.retry_with(stop=stop_after_attempt(4))()
except Exception:
    pass

print(raise_my_exception.statistics)

.. testoutput:: :hide:

...

如果你想使用变量来设置重试参数，你不必使用 retry 装饰器 - 你可以直接使用 Retrying:

.. testcode::

def never_good_enough(arg1):
    raise Exception('无效参数: {}'.format(arg1))

def try_never_good_enough(max_attempts=3):
    retryer = Retrying(stop=stop_after_attempt(max_attempts), reraise=True)
    retryer(never_good_enough, '我真的很努力')

你可能还想暂时更改装饰函数的行为，比如在测试中避免不必要的等待时间。你可以修改/补丁附加到函数的 retry 属性。请记住，这是一个只写属性，统计信息应该从函数的 statistics 属性读取。

.. testcode::

@retry(stop=stop_after_attempt(3), wait=wait_fixed(3))
def raise_my_exception():
    raise MyException("失败")

from unittest import mock

with mock.patch.object(raise_my_exception.retry, "wait", wait_fixed(0)):
    try:
        raise_my_exception()
    except Exception:
        pass

print(raise_my_exception.statistics)

.. testoutput:: :hide:

...

重试代码块

Tenacity 允许你重试代码块，而无需将其包装在独立的函数中。这使得在共享上下文的同时隔离失败的代码块变得容易。诀窍是结合使用 for 循环和上下文管理器。

.. testcode::

   from tenacity import Retrying, RetryError, stop_after_attempt

   try:
       for attempt in Retrying(stop=stop_after_attempt(3)):
           with attempt:
               raise Exception('我的代码失败了!')
   except RetryError:
       pass

你可以通过配置 Retrying 对象来配置重试策略的每个细节。

对于异步代码，你可以使用 AsyncRetrying。

.. testcode::

   from tenacity import AsyncRetrying, RetryError, stop_after_attempt

   async def function():
      try:
          async for attempt in AsyncRetrying(stop=stop_after_attempt(3)):
              with attempt:
                  raise Exception('我的代码失败了!')
      except RetryError:
          pass

在这两种情况下，你可能想将结果设置给 attempt，以便在像 ``retry_if_result`` 这样的重试策略中使用。这可以通过访问 ``retry_state`` 属性来完成：

.. testcode::

    from tenacity import AsyncRetrying, retry_if_result

    async def function():
       async for attempt in AsyncRetrying(retry=retry_if_result(lambda x: x < 3)):
           with attempt:
               result = 1  # 一些复杂的计算、函数调用等
           if not attempt.retry_state.outcome.failed:
               attempt.retry_state.set_result(result)
       return result

异步和重试
~~~~~~~~~~~~~~~

最后，``retry`` 也适用于 asyncio、Trio 和 Tornado (>= 4.5) 协程。睡眠也是异步完成的。

.. code-block:: python

    @retry
    async def my_asyncio_function(loop):
        await loop.getaddrinfo('8.8.8.8', 53)

.. code-block:: python

    @retry
    async def my_async_trio_function():
        await trio.socket.getaddrinfo('8.8.8.8', 53)

.. code-block:: python

    @retry
    @tornado.gen.coroutine
    def my_async_tornado_function(http_client, url):
        yield http_client.fetch(url)

你甚至可以通过传递正确的睡眠函数来使用其他事件循环，如 `curio`：

.. code-block:: python

    @retry(sleep=curio.sleep)
    async def my_async_curio_function():
        await asks.get('https://example.org')

贡献
----------
1. 检查是否有未解决的问题，或者开启一个新的问题，以讨论功能想法或Bug。
2. 在GitHub上fork`仓库`_，开始在**main**分支（或从其分出的分支）上进行修改。
3. 编写一个测试，以证明Bug已修复或功能按预期工作。
4. 添加一个`更新日志 <#更新日志>`_
5. 改进文档（使其更详细、更易读，或其他改进）

.. _`仓库`: https://github.com/jd/tenacity

更新日志
~~~~~~~~

使用`reno`_来管理更新日志。请查看其使用文档。

文档生成将自动编译更新日志。你只需要添加它们。

.. code-block:: sh

    # 在编辑器中打开一个模板文件
    tox -e reno -- new 我的更改的某个标识 --edit

.. _`reno`: https://docs.openstack.org/reno/latest/user/usage.html