当前位置:首页>python>stamina,一个舞动的python项目!

stamina,一个舞动的python项目!

  • 2026-10-11 10:40:33
stamina,一个舞动的python项目!

在写爬虫或调用第三方API时,我们经常遇到网络抖动或服务短暂不可用。

大多数人的解决方案是写个循环加time.sleep,但这往往让代码变得臃肿且不够健壮。

今天介绍的stamina库,基于Tenacity封装,提供了更符合直觉的重试机制。

🛠️ 基础用法:装饰器模式

stamina最优雅的地方在于它的声明式API。我们只需要在函数上添加装饰器,就能让函数具备智能重试能力,将业务逻辑与容错逻辑彻底分离。

import httpx
import 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 stamina

stamina.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通过简单的装饰器,解决了分布式系统中瞬态故障处理的三大痛点:重试次数、退避策略和异常过滤。

它让代码更简洁,也让系统更健壮。

你通常在项目中如何优雅地处理重试逻辑?欢迎在评论区分享你的“防崩溃”秘籍!

最新文章

随机文章