🚀 告别重复造轮子:Python stamina 模块优雅重试实战
在写爬虫或调用第三方API时,我们经常遇到网络抖动或服务短暂不可用。大多数人的解决方案是写个循环加time.sleep,但这往往让代码变得臃肿且不够健壮。今天介绍的stamina库,基于Tenacity封装,提供了更符合直觉的重试机制。
🛠️ 基础用法:装饰器模式
stamina最优雅的地方在于它的声明式API。我们只需要在函数上添加装饰器,就能让函数具备智能重试能力,将业务逻辑与容错逻辑彻底分离。
import httpximport stamina@stamina.retry(on=httpx.HTTPError, attempts=3)deffetch_data(url: str) -> dict: resp = httpx.get(url) resp.raise_for_status()return resp.json()
执行结果如下:
# 假设第一次请求超时Attempt 1 failed: HTTPError... 正在重试...# 等待0.1秒后重试(指数退避)Attempt 2 succeeded. 返回JSON数据。
🧠 高级控制:谓词函数与退避
很多时候我们只想重试特定的错误(比如5xx服务器错误),而不是所有的请求异常。stamina允许传入谓词函数进行精细控制,配合指数退避策略,能有效防止重试风暴压垮服务器。
defis_retryable(exc: Exception) -> bool:returnisinstance(exc, httpx.HTTPStatusError) and500 <= exc.response.status_code < 600@stamina.retry(on=is_retryable, attempts=5, wait_initial=1.0, wait_max=10.0)defunstable_task():print("执行核心任务...")raise httpx.HTTPStatusError("Server Error", response=..., request=...)
执行结果如下:
执行核心任务...捕获到服务器500错误,1秒后重试...执行核心任务...捕获到服务器500错误,2秒后重试...执行核心任务...成功!
🧪 测试支持与异步场景
在单元测试中,我们不想真的等待几秒钟。stamina提供了set_testing模式,可以全局禁用重试或移除退避等待,让测试瞬间完成。同时它也原生支持async/await语法。
import staminastamina.set_testing(True) # 禁用真实等待@stamina.retry(on=ValueError, attempts=3)deftest_func():raise ValueError("Bad Request")
执行结果如下:
# 测试环境下不再卡顿,快速失败抛出异常Traceback (most recent call last): ...ValueError: Bad Request
⚖️ 优势对比分析
相比Tenacity,stamina的优势在于“开箱即用”。Tenacity虽然强大但参数繁多,而stamina预设了合理的指数退避和抖动,减少了配置错误。其不足之处在于生态相对年轻,极端定制化场景下可能不如Tenacity灵活。建议:日常开发首选stamina,遇到特殊需求时再回退到底层Tenacity。
📝 总结
stamina通过简单的装饰器,解决了分布式系统中瞬态故障处理的三大痛点:重试次数、退避策略和异常过滤。它让代码更简洁,也让系统更健壮。
你通常在项目中如何优雅地处理重试逻辑?欢迎在评论区分享你的“防崩溃”秘籍!