stamina:轻松实现生产级重试
在分布式系统中,暂时性故障很常见。要使系统具有弹性,需要对失败的操作进行重试。但糟糕的重试可能会让情况变得更糟。
stamina是对优秀但缺乏主见的Tenacity包的一个有主见的封装。我们的目标是尽可能符合人体工程学,同时默认做正确的事,并最大限度地减少误用的可能性。它是多年来反复复制粘贴相同配置的结果:
- 仅对特定异常进行重试 - 甚至可以通过首先使用谓词检查异常的子集。
- 重试之间进行指数退避并加入抖动。
- 限制重试次数和总时间。
- 自动异步支持 - 包括Trio。
- 保留被装饰可调用对象的类型提示。
- 灵活的仪表支持,内置Prometheus、structlog和标准库的
logging
支持。 - 易于进行_全局_停用以进行测试。
例如:
import httpx
import stamina
@stamina.retry(on=httpx.HTTPError, attempts=3)
def do_it(code: int) -> httpx.Response:
resp = httpx.get(f"https://httpbin.org/status/{code}")
resp.raise_for_status()
return resp
异步可调用对象使用相同的API,也可以重试任意代码块。查看我们的教程以获取更多示例!
项目链接
致谢
stamina由Hynek Schlawack编写,并根据MIT许可证的条款分发。
开发工作得到了我的雇主Variomedia AG以及所有我出色的GitHub赞助者的大力支持。
如果没有Tenacity多年来的出色工作,这个项目是不可能实现的。