后端开发工具【免费下载链接】tenacityRetrying library for Python项目地址https://gitcode.com/gh_mirrors/te/tenacity点击查看免费下载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 模块文档区域对应模块提供的策略类型Retry Main APItenacity/init.pyretry装饰器、Retrying、AsyncRetrying、TornadoRetrying、RetryCallStateAfter Functionstenacity/after.pyafter关键字参数用Before Functionstenacity/before.pybefore关键字参数用Before Sleep Functionstenacity/before_sleep.pybefore_sleep关键字参数用Nap Functionstenacity/nap.pysleep关键字参数用Retry Functionstenacity/retry.pyretry关键字参数用Stop Functionstenacity/stop.pystop关键字参数用Wait Functionstenacity/wait.pywait关键字参数用所有内置策略都已在 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(stopstop_after_attempt(7)) def bar(): ... # 带参数写法其内部会按被装饰函数类型自动分派到三种控制器tenacity/init.py若函数是协程函数或sleep是协程函数→ 使用AsyncRetrying若函数是 Tornado 协程函数 → 使用TornadoRetrying其余普通函数 → 使用Retrying。retry的所有关键字参数最终都原样传递给对应控制器构造函数参数表见下文Retrying。2.tenacity.Retrying同步重试控制器Retrying是同步重试的核心控制器继承自抽象基类BaseRetrying。完整构造参数及默认值见 tenacity/init.py参数默认值说明sleeptenacity.nap.sleep即time.sleep每次重试间隔执行的睡眠函数stopstop_never停止策略决定何时放弃重试waitwait_none()等待策略决定重试前睡多久retryretry_if_exception_type()重试条件决定某次结果是否需要重试beforebefore_nothing每次尝试前执行的回调afterafter_nothing每次尝试结束后执行的回调before_sleepNone入睡前执行的回调如日志reraiseFalse放弃重试时是否直接抛出最后一次异常否则抛RetryErrorretry_error_clsRetryError重试耗尽后抛出的异常类retry_error_callbackNone重试耗尽时执行的回调返回其值代替抛异常nameNone重试对象名称用于日志与 reprenabledTrue设为False时跳过所有重试逻辑直接调用原函数控制器通过__call__驱动主循环tenacity/init.py每次迭代要么执行DoAttempt调用目标函数并记录结果/异常要么执行DoSleep睡眠后进入下一次尝试直到返回最终结果或抛RetryError。两种典型用法from tenacity import Retrying, stop_after_attempt # 方式一直接调用 result Retrying(stopstop_after_attempt(3), reraiseTrue)(fn, arg1, kw2) # 方式二作为迭代器/上下文管理器单次尝试粒度 for attempt in Retrying(stopstop_after_attempt(3)): with attempt: result fn(arg1)BaseRetrying还提供两个高频辅助能力wraps(f)把一个函数包装为带重试能力的装饰器包装结果会附带retry、retry_with、statistics属性tenacity/init.pycopy(...)在保留其余配置不变的前提下生成参数修改后的新控制器retry_with即基于它实现tenacity/init.py。3.tenacity.AsyncRetryingasyncio 重试控制器AsyncRetrying是Retrying的异步版本tenacity/asyncio/init.py参数签名与同步版完全一致但默认sleep是_portable_async_sleep——它会检测当前运行的是 trio 还是 asyncio 事件循环并调用trio.sleep或asyncio.sleeptenacity/asyncio/init.py。它同样支持异步上下文管理器__aiter__/__anext__与直接await调用两种形态from tenacity import AsyncRetrying, stop_after_attempt, wait_fixed async def main(): # 直接 await 调用 result await AsyncRetrying(stopstop_after_attempt(3), waitwait_fixed(1))(fn) # 异步迭代器形态 async for attempt in AsyncRetrying(stopstop_after_attempt(3)): with attempt: result await fn()注意AsyncRetrying.__iter__会抛出TypeErrorAsyncRetrying object is not iterable防止误用同步迭代tenacity/asyncio/init.py。异步侧还提供了独立的异步重试策略模块 tenacity/asyncio/retry.pyretry_if_exception、retry_if_result、retry_any、retry_all其谓词可以是awaitable。4.tenacity.tornadoweb.TornadoRetryingTornado 重试控制器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字段/方法类型含义start_timefloat重试开始时间戳time.monotonic()retry_objectBaseRetrying当前重试管理器fn/args/kwargs—被重试的函数及其参数attempt_numberint当前尝试次数从 1 开始outcomeFuture | None最近一次结果或异常Future.failed判断是否异常outcome_timestampfloat最近一次结果的时间戳idle_forfloat累计睡眠时间next_action/upcoming_sleep—下一步动作与即将执行的睡眠时长get_fn_name()str被重试函数的全限定名用于日志seconds_since_startfloat距首次尝试经过的秒数自定义策略示例判断耗时是否超过阈值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.06.RetryError、TryAgain与运行统计RetryError封装放弃前最后一次尝试的Future通过retry_error_callback或reraise控制抛出方式tenacity/init.pyTryAgain是一个特殊异常在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 Functionsafter参数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(levellogging.DEBUG) retry(stopstop_after_attempt(3), afterafter_log(logger, logging.DEBUG)) def fetch(): ...四、Before Functionsbefore参数before回调在每次尝试前执行模块 tenacity/before.py 提供before_nothing(retry_state)什么都不做默认值before_log(logger, log_level)记录即将开始第 N 次调用内部同样依赖retry_state.get_fn_name()与attempt_numbertenacity/before.py。五、Before Sleep Functionsbefore_sleep参数before_sleep在确认需要重试、且睡眠开始之前执行是记录为什么重试、多久后重试的最佳位置。模块 tenacity/before_sleep.py 提供before_sleep_nothing(retry_state)什么都不做before_sleep_log(logger, log_level, exc_infoFalse, sec_format%.3g)记录如Retrying fetch in 1.0 seconds as it raised ConnectionError: ...这样的日志exc_infoTrue时额外附带完整 tracebacktenacity/before_sleep.py。注意该回调要求outcome与next_action均已设置即只应在重试循环内部使用。retry(stopstop_after_attempt(3), before_sleepbefore_sleep_log(logger, logging.WARNING, exc_infoTrue)) def connect(): ...六、Nap Functionssleep参数sleep决定如何真正入睡模块 tenacity/nap.py 提供sleep(seconds)默认策略直接调用time.sleep(seconds)。文档注释特别说明它可以被 mock 以便单元测试tenacity/nap.pysleep_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(stopstop_after_attempt(3), sleepfake_sleep)七、Retry Functionsretry参数retry策略决定某次尝试结果是否值得重试其输入是RetryCallState返回bool。模块 tenacity/retry.py 以retry_base为抽象基类其__call__返回bool并支持AND与|OR运算符组合——retry_all与retry_anytenacity/retry.py。策略签名与默认值行为retry_never单例永不重试恒 Falseretry_always单例总是重试恒 True需配合 stop 防止死循环retry_if_exception(predicate)谓词BaseException - bool异常满足谓词则重试retry_if_exception_type(exception_typesException)类型或类型元组抛出的异常是指定类型含子类则重试默认捕获所有Exceptionretry_if_not_exception_type(exception_typesException)同上抛出的异常不是指定类型则重试retry_unless_exception_type(exception_typesException)同上未抛异常或异常不是指定类型时都重试直到抛出指定类型才停止retry_if_exception_cause_type(exception_typesException)同上沿__cause__链递归检查异常原因是否为指定类型且能识别循环链如raise e from eretry_if_result(predicate)谓词result - bool返回值满足谓词则重试retry_if_not_result(predicate)谓词result - bool返回值不满足谓词则重试retry_if_exception_message(messageNone, matchNone)二选一异常消息等于message或用match字符串或已编译正则匹配两者都传会抛TypeErrorretry_if_not_exception_message(messageNone, matchNone)二选一与上相反消息不匹配时重试retry_any(*retries)任意个任一子条件为真则重试retry_all(*retries)任意个全部子条件为真才重试典型用法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(retryretry_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(retryretry_unless_exception_type(Timeout)) def poll_until_timeout(): ...retry_if_exception_cause_type用于处理异常被包装的场景如网络库把真实原因放在__cause__里实现见 tenacity/retry.py。八、Stop Functionsstop参数stop策略决定何时整体放弃重试输入RetryCallState返回bool。模块 tenacity/stop.py 提供策略签名行为stop_never单例永不停止配合retry_always会无限循环需谨慎stop_after_attempt(max_attempt_number)int尝试次数 ≥max_attempt_number时停止stop_after_delay(max_delay)时间自首次尝试起耗时 ≥max_delay时停止文档明确提示实际总延迟可能略超上限因为会先执行完最后一次等待需要严格控时请用stop_before_delaystop_before_delay(max_delay)时间在当前已耗时 即将到来的睡眠 ≥max_delay时停止即保证不超出上限适合配合随机/指数等待使用stop_when_event_set(event)threading.Event事件被 set 时停止可实现外部取消stop_any(*stops)/stop_all(*stops)任意个任一满足 / 全部满足才停止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(stopstop_after_attempt(7)) def a(): ... retry(stopstop_after_delay(10)) # 最多重试 10 秒可能略超 def b(): ... retry(stopstop_before_delay(10)) # 严格不超过 10 秒 def c(): ... retry(stop(stop_after_delay(10) | stop_after_attempt(5))) # 先到先停 def d(): ...九、Wait Functionswait参数wait策略决定每次重试前等待多久输入RetryCallState返回float。模块 tenacity/wait.py 提供策略签名与默认值行为wait_none()—不等待立即重试默认值wait_fixed(wait)时间每次固定等待waitwait_random(min0, max1)时间在[min, max]间均匀随机wait_incrementing(start0, increment100, maxMAX_WAIT)时间每次递增start increment * (attempt-1)并截断到[0, max]wait_exponential(multiplier1, maxMAX_WAIT, exp_base2, min0)时间指数退避multiplier * exp_base ** (attempt-1)夹在[min, max]内无抖动适合资源暂时不可用、而非多进程争抢的场景wait_random_exponential(multiplier1, maxMAX_WAIT, exp_base2, min0)时间在[min, 指数上限]区间内均匀随机实现Full Jitter式退避适合多进程争抢共享资源的场景wait_exponential_jitter(initial1, maxMAX_WAIT, exp_base2, jitter1, min0, multiplier1)时间max(min, min(multiplier * 2**n uniform(0, jitter), max))initial已弃用传它且同时传multiplier会抛ValueError请统一使用multiplierwait_chain(*strategies)至少一个按尝试次数顺序切换等待策略全部用尽后沿用最后一个wait_combine(*strategies)任意个每次等待 所有子策略返回值之和wait_exception(predicate)谓词exception - float依据异常对象动态决定等待时长例如读取 HTTPRetry-After响应头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(waitwait_fixed(2)) def a(): ... retry(waitwait_random(min1, max2)) def b(): ... retry(waitwait_exponential(multiplier1, min4, max10)) def c(): ... retry(waitwait_fixed(3) wait_random(0, 2)) # 组合固定 3s 随机 0~2s def d(): ... retry(waitwait_random_exponential(multiplier0.5, max60)) def e(): ... # 前 3 次等 1s接着 5 次等 2s之后一直等 5s retry(waitwait_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(stopstop_after_attempt(3), waitwait_exception(http_error)) def rate_limited_call(): ...MAX_WAIT定义为sys.maxsize / 2tenacity/_utils.py即等待时长默认几乎无上限。wait_exponential在计算溢出时OverflowError会直接返回max不会崩溃tenacity/wait.py。十、组合策略的运算符约定三类策略分别定义了直观的运算符组合见 tenacity/retry.py、tenacity/stop.py、tenacity/wait.pyretry策略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.pyTornado 控制器可进一步验证各策略的实际行为。十一、把 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(levellogging.INFO) attempt 0 retry( stopstop_after_attempt(5), waitwait_exponential(multiplier1, min1, max8), retryretry_if_exception_type(ConnectionError) | retry_if_result(lambda r: r is None), before_sleepbefore_sleep_log(logger, logging.WARNING), afterafter_log(logger, logging.INFO), reraiseTrue, ) 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 结构的完整价值。赞分享后端开发工具【免费下载链接】tenacityRetrying library for Python项目地址https://gitcode.com/gh_mirrors/te/tenacity点击查看免费下载相关推荐Tenacity 重试库完全指南从 retry 装饰器到异步重试的 Python 弹性编程Tenacity 重试库完全指南从 retry 装饰器到异步重试的 Python 弹性编程 Tenacity 是一个 Apache 2.0 许可、用 Pyt后端开发工具探索Android开源世界从零开始构建你的第一个应用探索Android开源世界从零开始构建你的第一个应用 你是否曾梦想开发一款属于自己的Android应用却被复杂的技术栈和庞大的代码库吓退 今天我将带文档移动开发知识库openage Modding API 参考engine.modifier 修饰器模块完全指南openage Modding API 参考engine.modifier 修饰器模块完全指南 engine.modifier 是 openage 模组 AP游戏开发图形学上一篇Spring Data Relational 项目教程下一篇SFSafeSymbols架构设计可扩展符号系统的实现原理创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考