用于回退和重试的函数装饰器——现代、快速、零依赖。backon 是对 backoff 的现代进化——一个用于重试的零依赖 Python 库，带有指数退避。它为同步和异步代码提供了装饰器、函数式和上下文管理器 API。 目录 特性 安装 快速入门 API 参考 装饰器 函数式 API 上下文管理器 调用者 等待生成器 停止条件 重试条件 抖动 处理器 全局切换 异步支持 自定义睡眠 高级特性 电路断路器 对冲 指标 测试 工具 Trio 支持 重试上下文检查 动态回退 热循环检测 重试统计 操作符组合 迭代器 API 从 backoff 迁移 贡献 许可证 特性 零依赖——纯 Python，仅标准库 四种 API——装饰器（@on_exception，@on_predicate），函数式（retry()），上下文管理器（Retrying），可调用（RetryingCaller / AsyncRetryingCaller） 原生异步——相同的 API 适用于 async def 函数 完整类型提示——通过 mypy 校验，兼容严格模式 全局切换——backon.disable() / backon.enable() 用于测试 自定义睡眠——注入您自己的睡眠函数（对使用 asyncio.Event 进行测试很有用） 多种等待策略——指数、常量、斐波那契、衰减、运行时、随机、递增和可组合链 抖动——完全抖动、随机抖动或无 报告回调——on_attempt、on_backoff、on_success、on_giveup、before_sleep、before、after 电路断路器——CLOSED/OPEN/HALF_OPEN 状态及自动恢复 对冲——并发重试请求，先成功者获胜 Prometheus / OpenTelemetry 指标——可选，零硬依赖 测试模块——disable_retries()、limit_retries()、remove_backoff()、assert_retried() Trio 支持——使用 trio 异步框架重试 操作符重载——通过 | / & 组合停止，通过 + 组合等待生成器 迭代器 API——在 Retrying(...) 中 for attempt: 现代打包——PEP 621，PDM，py.typed 安装 pip install backon 需要 Python 3.10+。 快速入门 在异常上重试 import backon @backon.on_exception(backon.expo, ValueError, max_tries=3) def fetch_data(): return api.call() 在谓词上重试 @backon.on_predicate(backon.constant, max_tries=5, interval=0.5) def poll_status(): return check_ready() 函数式 API result = backon.retry(fetch_data, backon.expo, exception=ValueError, max_tries=3) 使用 backon.Retrying 的上下文管理器（backon.expo，exception=ValueError，max_tries=3）作为 r: result = r.call(fetch_data) 异步变体：async with backon.Retrying(backon.constant, exception=ValueError, max_tries=3, interval=0.5) as r: result = await r.async_call(fetch_data) API 参考 装饰器 @backon.on_exception(wait_gen, exception, ...) 装饰的函数在引发指定异常之一时重试。 @backon.on_exception(backon.expo, (ValueError, TimeoutError), max_tries=5) def fetch(): ... 参数 类型 默认值 描述 wait_gen WaitGenerator——等待策略（expo, constant, fibo 等） exception type 或 tuple[type]——要重试的异常类 max_tries int 或 Callable[[], int] None 最大尝试次数 max_time float，timedelta 或 Callable None 最大总花费时间 jitter Jitterer 或 None full_jitter 抖动函数 giveup Callable[[Exception], bool 或 float] lambda e: False 对于匹配的异常停止重试；返回 float 覆盖等待 on_success Handler 或 list None 在成功尝试后调用 on_backoff Handler 或 list None 在每次重试之前调用 on_giveup Handler 或 list None 当重试耗尽时调用 on_attempt Handler 或 list None 在每次尝试之前调用 before_sleep Handler 或 list None 在休眠之前调用 before Handler 或 list None 在每次尝试之前调用（低于 on_attempt） after Handler 或 list None 在每次尝试之后调用（低于 on_success/on_giveup） retry_error_callback Callable[[dict], Any] None 在重试放弃而不是引发时调用 raise_on_giveup bool True 当放弃时引发最终异常 logger str 或 Logger "backon" 日志名称或实例 backoff_log_level int logging.INFO 回退消息的日志级别 giveup_log_level int logging.ERROR 放弃消息的日志级别 sleep Callable[[float], Any] None 自定义睡眠函数 **wait_gen_kwargs varies——传递给等待生成器的额外 kwargs（例如 base=3，interval=0.5） @backon.on_predicate(wait_gen, predicate, ...) 在谓词匹配返回值时重试。 @backon.on_predicate(backon.constant, predicate=lambda x: x is None, max_tries=5) def poll(): ... 接受所有来自 on_exception 的参数，除了 exception，giveup 和 raise_on_giveup。 添加： 参数 类型 默认值 描述 predicate Callable[[Any], bool] operator.not_ 当其返回 True 时重试