tenacity Python 重试库完整 API 参考：retry 装饰器、Retrying 控制器与六大策略模块实战指南

• 后端
• 开发工具

tenacity 是一个为 Python 提供通用重试能力的开源库，其核心价值在于把"何时重试（retry）"、"何时停止（stop）"、"等待多久（wait）"、"重试前后做什么（before/after/before_sleep）"以及"如何睡眠（sleep/nap）"全部解耦为独立可组合的策略对象，并通过@retry装饰器与Retrying/AsyncRetrying/TornadoRetrying控制器统一驱动。本文以仓库文档 doc/source/api.rst 定义的 API 结构为骨架，逐一讲解主 API 与六大策略模块的每个公开符号、参数默认值与底层实现，读者可以据此掌握 tenacity 的全部公开接口，并直接写出可运行的同步、asyncio 与 Tornado 重试代码。

一、API 文档结构总览：一张 API 地图

doc/source/api.rst 把十点重试库的公开接口划分为 8 个区域，每个区域对应一个真实的 Python 模块：

所有内置策略都已在 tenacity/init.py 中集中导入，因此实践中可以直接从tenacity顶层命名空间导入，无需深入子模块。此外 tenacity/_utils.py 定义了time_unit_type = int | float | timedelta与to_seconds()转换函数——所有接受时间类参数的位置（wait 的时长、stop 的延迟上限等）都同时支持秒数（int/float）与datetime.timedelta对象，这是全库统一的约定。

二、Retry Main API：重试主入口

1.tenacity.retry：装饰器入口

retry是文档中最先出现的函数，也是最常用的入口。它同时支持@retry与@retry(...)两种写法，见 tenacity/init.py 的实现：

@retry def foo(): ... # 无参数写法 @retry(stop=stop_after_attempt(7)) def bar(): ... # 带参数写法

其内部会按被装饰函数类型自动分派到三种控制器（tenacity/init.py）：

• 若函数是协程函数（或sleep是协程函数）→ 使用AsyncRetrying；
• 若函数是 Tornado 协程函数 → 使用TornadoRetrying；
• 其余普通函数 → 使用Retrying。

retry的所有关键字参数最终都原样传递给对应控制器构造函数，参数表见下文Retrying。

2.tenacity.Retrying：同步重试控制器

Retrying是同步重试的核心控制器，继承自抽象基类BaseRetrying。完整构造参数及默认值见 tenacity/init.py：

控制器通过__call__驱动主循环（tenacity/init.py）：每次迭代要么执行DoAttempt（调用目标函数并记录结果/异常），要么执行DoSleep（睡眠后进入下一次尝试），直到返回最终结果或抛RetryError。两种典型用法：

from tenacity import Retrying, stop_after_attempt # 方式一：直接调用 result = Retrying(stop=stop_after_attempt(3), reraise=True)(fn, arg1, kw=2) # 方式二：作为迭代器/上下文管理器（单次尝试粒度） for attempt in Retrying(stop=stop_after_attempt(3)): with attempt: result = fn(arg1)

BaseRetrying还提供两个高频辅助能力：

• wraps(f)：把一个函数包装为带重试能力的装饰器，包装结果会附带retry、retry_with、statistics属性（tenacity/init.py）；
• copy(...)：在保留其余配置不变的前提下生成参数修改后的新控制器，retry_with即基于它实现（tenacity/init.py）。

3.tenacity.AsyncRetrying：asyncio 重试控制器

AsyncRetrying是Retrying的异步版本（tenacity/asyncio/init.py），参数签名与同步版完全一致，但默认sleep是_portable_async_sleep——它会检测当前运行的是 trio 还是 asyncio 事件循环，并调用trio.sleep或asyncio.sleep（tenacity/asyncio/init.py）。它同样支持异步上下文管理器（__aiter__/__anext__）与直接await调用两种形态：

from tenacity import AsyncRetrying, stop_after_attempt, wait_fixed async def main(): # 直接 await 调用 result = await AsyncRetrying(stop=stop_after_attempt(3), wait=wait_fixed(1))(fn) # 异步迭代器形态 async for attempt in AsyncRetrying(stop=stop_after_attempt(3)): with attempt: result = await fn()

注意：AsyncRetrying.__iter__会抛出TypeError（"AsyncRetrying object is not iterable"），防止误用同步迭代（tenacity/asyncio/init.py）。异步侧还提供了独立的异步重试策略模块 tenacity/asyncio/retry.py（retry_if_exception、retry_if_result、retry_any、retry_all，其谓词可以是awaitable）。

4.tenacity.tornadoweb.TornadoRetrying：Tornado 重试控制器

TornadoRetrying面向基于tornado.gen的协程（tenacity/tornadoweb.py），默认sleep为tornado.gen.sleep，其__call__以@gen.coroutine实现，通过yield等待目标函数与睡眠。仅在使用 Tornado 的代码中需要，tenacity在导入时会检测 tornado 是否可用（tenacity/init.py）。

5.tenacity.RetryCallState：单次调用的状态载体

RetryCallState是贯穿所有策略的对象——每个retry/stop/wait/before/after/before_sleep回调都会收到它，因此掌握其字段是编写自定义策略的前提。完整字段见 tenacity/init.py：

自定义策略示例（判断耗时是否超过阈值）：

from tenacity import RetryCallState def wait_if_slow(retry_state: RetryCallState) -> float: return 5.0 if retry_state.seconds_since_start and retry_state.seconds_since_start > 2 else 0.0

6.RetryError、TryAgain与运行统计

• RetryError封装"放弃前最后一次尝试"的Future，通过retry_error_callback或reraise控制抛出方式（tenacity/init.py）；
• TryAgain是一个特殊异常：在except块中主动raise TryAgain可无条件触发下一次重试（tenacity/init.py），且RetryError.reraise()会还原其底层原因异常；
• statistics属性返回运行期统计字典，典型键为start_time、attempt_number、idle_for、delay_since_first_attempt（由begin()初始化，见 tenacity/init.py）。统计是按线程隔离的（threading.local），多线程共享同一控制器时各自独立。

三、After Functions（after参数）

after回调在每次尝试结束后执行，返回值为空。模块 tenacity/after.py 提供两个内置实现：

• after_nothing(retry_state)：什么都不做（默认值）；
• after_log(logger, log_level, sec_format="%.3g")：以指定日志器与级别记录"本次调用耗时、这是第几次调用"，sec_format控制耗时格式化。底层调用retry_state.get_fn_name()获取函数名、retry_state.seconds_since_start获取耗时（tenacity/after.py）。

import logging from tenacity import retry, stop_after_attempt, after_log logger = logging.getLogger(__name__) logging.basicConfig(level=logging.DEBUG) @retry(stop=stop_after_attempt(3), after=after_log(logger, logging.DEBUG)) def fetch(): ...

四、Before Functions（before参数）

before回调在每次尝试前执行，模块 tenacity/before.py 提供：

• before_nothing(retry_state)：什么都不做（默认值）；
• before_log(logger, log_level)：记录"即将开始第 N 次调用"，内部同样依赖retry_state.get_fn_name()与attempt_number（tenacity/before.py）。

五、Before Sleep Functions（before_sleep参数）

before_sleep在确认需要重试、且睡眠开始之前执行，是记录"为什么重试、多久后重试"的最佳位置。模块 tenacity/before_sleep.py 提供：

• before_sleep_nothing(retry_state)：什么都不做；
• before_sleep_log(logger, log_level, exc_info=False, sec_format="%.3g")：记录如Retrying fetch in 1.0 seconds as it raised ConnectionError: ...这样的日志；exc_info=True时额外附带完整 traceback（tenacity/before_sleep.py）。注意该回调要求outcome与next_action均已设置，即只应在重试循环内部使用。

@retry(stop=stop_after_attempt(3), before_sleep=before_sleep_log(logger, logging.WARNING, exc_info=True)) def connect(): ...

六、Nap Functions（sleep参数）

sleep决定"如何真正入睡"，模块 tenacity/nap.py 提供：

• sleep(seconds)：默认策略，直接调用time.sleep(seconds)。文档注释特别说明它可以被 mock 以便单元测试（tenacity/nap.py）；
• sleep_using_event(event)：返回一个等待threading.Event被 set 的睡眠函数，事件一旦被设置会提前结束等待（event.wait(timeout)），适合用于实现可中断/可取消的重试（tenacity/nap.py）。

自定义sleep也很常见，例如写入测试中让nap.sleep可被 mock 后立即返回：

from tenacity import Retrying, stop_after_attempt import tenacity.nap def fake_sleep(seconds): pass # 测试中不真正阻塞 retryer = Retrying(stop=stop_after_attempt(3), sleep=fake_sleep)

七、Retry Functions（retry参数）

retry策略决定某次尝试结果是否值得重试，其输入是RetryCallState，返回bool。模块 tenacity/retry.py 以retry_base为抽象基类（其__call__返回bool），并支持&（AND）与|（OR）运算符组合——retry_all与retry_any（tenacity/retry.py）。

典型用法：

from tenacity import ( retry, retry_if_exception_type, retry_if_not_exception_type, retry_if_result, retry_unless_exception_type, ) from requests import ConnectionError, Timeout # 只重试特定异常 @retry(retry=retry_if_exception_type((ConnectionError, Timeout))) def call_api(): ... # 组合：连接错误重试，且返回值为 None 也重试 @retry(retry=(retry_if_result(lambda r: r is None) | retry_if_exception_type())) def get_user(): ... # 直到出现特定异常才停止重试（常用于"等到某错误发生"） @retry(retry=retry_unless_exception_type(Timeout)) def poll_until_timeout(): ...

retry_if_exception_cause_type用于处理"异常被包装"的场景（如网络库把真实原因放在__cause__里），实现见 tenacity/retry.py。

八、Stop Functions（stop参数）

stop策略决定何时整体放弃重试，输入RetryCallState，返回bool。模块 tenacity/stop.py 提供：

stop_base同样支持&（stop_all）与|（stop_any）运算符（tenacity/stop.py）。所有*_delay的max_delay都接受 int/float/timedelta，内部经_utils.to_seconds归一化（tenacity/stop.py）。

from tenacity import retry, stop_after_attempt, stop_after_delay, stop_before_delay @retry(stop=stop_after_attempt(7)) def a(): ... @retry(stop=stop_after_delay(10)) # 最多重试 10 秒（可能略超） def b(): ... @retry(stop=stop_before_delay(10)) # 严格不超过 10 秒 def c(): ... @retry(stop=(stop_after_delay(10) | stop_after_attempt(5))) # 先到先停 def d(): ...

九、Wait Functions（wait参数）

wait策略决定每次重试前等待多久，输入RetryCallState，返回float。模块 tenacity/wait.py 提供：

wait_base支持+运算符（等价于wait_combine），并且实现了__radd__，因此多个等待策略可以直接用内置sum()相加（tenacity/wait.py）。

from tenacity import ( retry, wait_fixed, wait_random, wait_exponential, wait_random_exponential, wait_chain, wait_exception, ) @retry(wait=wait_fixed(2)) def a(): ... @retry(wait=wait_random(min=1, max=2)) def b(): ... @retry(wait=wait_exponential(multiplier=1, min=4, max=10)) def c(): ... @retry(wait=wait_fixed(3) + wait_random(0, 2)) # 组合：固定 3s + 随机 0~2s def d(): ... @retry(wait=wait_random_exponential(multiplier=0.5, max=60)) def e(): ... # 前 3 次等 1s，接着 5 次等 2s，之后一直等 5s @retry(wait=wait_chain(*[wait_fixed(1) for _ in range(3)] + [wait_fixed(2) for _ in range(5)] + [wait_fixed(5) for _ in range(4)])) def f(): ... # 依据异常动态等待（示例来自 wait_exception 的 docstring） def http_error(exception): if isinstance(exception, requests.HTTPError) and exception.response.status_code == 429: return float(exception.response.headers.get("Retry-After", "1")) return 60.0 @retry(stop=stop_after_attempt(3), wait=wait_exception(http_error)) def rate_limited_call(): ...

MAX_WAIT定义为sys.maxsize / 2（tenacity/_utils.py），即等待时长默认几乎无上限。wait_exponential在计算溢出时（OverflowError）会直接返回max，不会崩溃（tenacity/wait.py）。

十、组合策略的运算符约定

三类策略分别定义了直观的运算符组合（见 tenacity/retry.py、tenacity/stop.py、tenacity/wait.py）：

• retry策略：A & B→ 全部满足才重试；A | B→ 任一满足即重试；
• stop策略：A & B→ 全部满足才停止；A | B→ 任一满足即停止；
• wait策略：A + B→ 每次等待两者之和；sum([wait_fixed(1), wait_random(0, 1)])也可用。

这套运算符使十点重试库的策略可以像搭积木一样组合，README 中大量示例均依赖此约定，仓库测试 tests/test_tenacity.py（同步策略与主控制器）、tests/test_asyncio.py（异步控制器与 trio/asyncio 兼容）、tests/test_tornado.py（Tornado 控制器）可进一步验证各策略的实际行为。

十一、把 API 串联起来：一个完整可运行示例

综合以上全部 API，一个兼顾异常重试、结果校验、指数退避、日志与统计的完整示例：

import logging import time from tenacity import ( retry, retry_if_exception_type, retry_if_result, stop_after_attempt, wait_exponential, before_sleep_log, after_log, ) logger = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO) attempt = 0 @retry( stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=1, max=8), retry=retry_if_exception_type(ConnectionError) | retry_if_result(lambda r: r is None), before_sleep=before_sleep_log(logger, logging.WARNING), after=after_log(logger, logging.INFO), reraise=True, ) def unstable_call(): global attempt attempt += 1 if attempt < 3: raise ConnectionError("temporary failure") return None # 第三次也返回 None，触发结果重试

运行后可以看到：前两次因ConnectionError重试、第三次因返回None重试，每次入睡前打印Retrying ... in X seconds as it raised ...，每次尝试结束打印耗时日志，最终第 5 次成功返回或抛RetryError/原异常。读者可按需替换stop/wait/retry/before/after/before_sleep/sleep中的任何一个为自定义回调（只需符合RetryCallState输入约定），即可无限扩展十点重试库的行为，这正是 doc/source/api.rst 所定义 API 结构的完整价值。

• 后端
• 开发工具