当前位置：首页 > news >正文

[python3] 处理函数的重试

news 2025/7/18 6:39:48

tenacity是一个 Python 库，用于简化重试逻辑的实现。它提供了装饰器和工具函数，使得在函数执行失败时可以自动重试。以下是对tenacity库的详细介绍：

一、安装

可以使用pip安装tenacity：

pip install tenacity

二、主要概念和功能

装饰器：
- @retry：这是最常用的装饰器，用于在函数执行失败时自动重试。可以通过参数配置重试的条件、次数、延迟等。
- @retry_always：无论函数执行结果如何，始终重试。
- @retry_if_exception_type：根据异常类型决定是否重试。例如，可以指定只在特定类型的异常发生时重试。
重试条件：
- retry(wait=...)：指定重试之间的等待时间。可以使用内置的等待策略，如指数退避（wait_exponential）、固定等待时间（wait_fixed）等。
- retry(stop=...)：定义重试的停止条件。可以设置最大重试次数（stop_max_attempt_number）、最大等待时间（stop_max_delay）等。
- retry(retry=...)：自定义重试的条件函数。该函数接收一个参数，表示上一次执行的结果，如果返回True则重试，否则停止。
异常处理：
- retry_if_exception_type(...)：指定在特定类型的异常发生时重试。可以传入一个或多个异常类型。
- retry_if_not_result(...)：根据函数的返回值决定是否重试。例如，可以检查返回值是否为None或满足其他条件。
回调函数：
- retry(after=...)：在每次重试后执行一个回调函数。可以用于记录日志、发送通知等。
- retry(before=...)：在每次重试前执行一个回调函数。可以用于准备资源、设置状态等。

三、使用示例

基本重试：

import tenacity
import random@tenacity.retry
def some_function():result = random.randint(1, 10)if result < 5:raise Exception("Failed.")else:return resulttry:print(some_function())
except Exception as e:print(f"Final failure: {e}")

在这个例子中，some_function会在返回值小于 5 时抛出异常并自动重试，直到返回值大于等于 5。

自定义重试条件和等待时间：

import tenacity
import timedef is_failure(result):return result is False@tenacity.retry(wait=tenacity.wait_fixed(2), retry=tenacity.retry_if_result(is_failure))
def another_function():result = random.choice([True, False])print(f"Result: {result}")return resulttry:another_function()
except Exception as e:print(f"Final failure: {e}")

在这个例子中，another_function会在返回值为False时重试，每次重试之间等待 2 秒。自定义的重试条件函数is_failure根据返回值决定是否重试。

使用回调函数：

import tenacity
import loggingdef log_retry(retry_state):logging.info(f"Retrying {retry_state.fn.__name__}, attempt #{retry_state.attempt_number}")@tenacity.retry(wait=tenacity.wait_exponential(multiplier=1, min=2, max=10), after=log_retry)
def yet_another_function():result = random.randint(1, 10)if result < 5:raise Exception("Failed.")else:return resulttry:yet_another_function()
except Exception as e:print(f"Final failure: {e}")