您现在的位置是：网站首页> 编程资料编程资料

Python重试库 Tenacity详解(推荐)_python_

2023-05-26 398人已围观

简介 Python重试库 Tenacity详解(推荐)_python_

1 Tenacity描述

今天 给大家介绍一个Python 重试库，Tenacity 这个库 是我 这些年 使用的一个非常好的库，几乎满足了我所有的重试需求，非常棒的一个库。

在我们日常开发中，我们经常会遇到 网络问题 或者其他问题，导致 API请求失败，或者超时等出现， 这个出现是偶发性，不确定性的，我们如何保证系统的功能的正确性呢？ 第一种 显示 告诉 用户请求失败。 第二种 我们可以通过重试 几次 看看会不会成功，如果 能成功 就可以了。

因此在真实的项目开发中重试的库，也显得 尤其重要， 在工作中 我也有尝试写过一些 重试的装饰器函数，总感觉 差点意思，要不就是 考虑不全面，要么就是 实现太简单 ，不能针对具体的异常进行重试，不能设置一些条件重试等。 以及重试 还是失败了怎么处理？ 等等一系列的问题。

直到后来发现这个 宝藏库 tenacity ，几乎解决了我所有的痛点。

一起来了解这个库的基本用法

2 如果发生异常就重试

这个就是 如果发生了异常就一直重试下去：

from tenacity import retry, stop_after_attempt @retry() def never_gonna_give_you_up(): print("Retry forever ignoring Exceptions, don't wait between retries") raise Exception("Unknown Error")

3 设置停止重试的条件

设置重试的最大次数

这里设置最多重试5次 ，失败后 就不在进行重试了。

from tenacity import retry, stop_after_attempt @retry(stop=stop_after_attempt(5)) def never_gonna_give_you_up(): print("Retry forever ignoring Exceptions, don't wait between retries") raise Exception("Unknown Error") if __name__ == '__main__': never_gonna_give_you_up()

还可以设置stop 时间

stop = stop_after_delay

Stop when the time from the first attempt >= limit 从第一次失败 尝试 到 多久后 停止尝试。

默认 stop_after_delay 单位是秒

import time from tenacity import retry, stop_after_delay @retry(stop=stop_after_delay(10)) def stop_after_10_s(): print("10秒后停止重试") time.sleep(5) raise Exception("Unknown Error") if __name__ == '__main__': stop_after_10_s()

停止重试条件 进行组合

from tenacity import retry, stop_after_delay, stop_after_attempt @retry(stop=(stop_after_delay(10) | stop_after_attempt(5))) def stop_after_10_s_or_5_retries(): print("10秒后或者5次重试后停止重试") raise Exception("Unknown Error") if __name__ == '__main__': stop_after_10_s_or_5_retries()

4 设置重试的间隔

有的时候 重试 太频繁了，也不太好，可能因为 网络问题，过一段时间才能恢复好，

wait_fixed(seconds=10) 等几秒后重试， 然后 可以组合 stop 在一起使用。

from tenacity import retry, stop_after_attempt, wait_fixed @retry(stop=stop_after_attempt(3),wait=wait_fixed(3)) def never_gonna_give_you_up(): print("Retry max attempt:3, wait_fixed:3") raise Exception("Unknown Error") if __name__ == '__main__': never_gonna_give_you_up()

5 重试触发的条件

针对具体的异常进行重试

对具体的异常进行重试 ,而不是所有异常进行重试. 有的时候 我们在写 业务代码的时候，可能会定义一些 自定义的异常，我们有时候 触发特定的异常 进行重试，而不是说 发生了异常 就进行 重试。 tenacity 这个库 早就想好了这点。

我自定义了一个异常 NotExistResourceErr ， 触发这个异常我才进行重试，重试最多三次。

# -*- coding: utf-8 -*- from typing import Any from tenacity import retry, stop_after_delay, stop_after_attempt, retry_if_exception_type class NotExistResourceErr(Exception): """ 定义一个自定义的异常 """ code = str(400) detail = "Parameter Error" def __init__(self, code: str = None, detail: Any = None): self.code = code self.detail = detail class Source: def __init__(self, id): self.id = id @retry(retry=retry_if_exception_type(NotExistResourceErr), stop=stop_after_attempt(3)) def might_exist_error(task_id: int = 0): print('might_exist_error begin ...') if task_id < 0: raise NotExistResourceErr() return Source(task_id) if __name__ == '__main__': might_exist_error(-10)

retry_if_exception_type 这里就告诉 tenacity 在发生什么类型，才进行重试操作。 设置 stop_after_attempt(3) 重试3 次就停止了。

运行一下结果如下： 可以看出 重试了三次，第三次后 抛出了异常

如果在里面 这样 抛出一个异常的话，此时 tenacity 不会进行重试操作，而是 程序直接报错。

@retry(retry=retry_if_exception_type(NotExistResourceErr), stop=stop_after_attempt(3)) def might_exist_error(task_id: int = 0): print('might_exist_error begin ...') if task_id < 0: raise Exception("Unknown Error") return Source(task_id)

如果这样的话， 不会进行重试操作 ，而是直接抛出了异常。

针对函数的返回结果进行重试

有的时候 可能因为 某些原因 正常情况下，应该可以获取到值，但是返回结果为None, 异常情况 在 函数内部进行捕获处理了，这个时候 也想进行 重试 怎么办呢？

retry_if_result 这里可以传入一个函数，这个函数接收 might_return_none 返回值， 然后 这个函数 返回 True 就会 触发重试操作。

from tenacity import retry, stop_after_attempt, retry_if_result def is_none(value): """Return True if value is None""" return value is None @retry(retry=retry_if_result(is_none), stop=stop_after_attempt(3)) def might_return_none(): print("返回 None 则重试") return None if __name__ == '__main__': might_return_none()

类似与此相反的 retry_if_not_result 当结果 不符合预期 则进行重试操作。

6 定义重试失败回调函数

对于重要的业务的话， 如果重试几次后， 发现仍然是失败的状态，此时 我们可以 设置 一个回调函数， 比如 在回调函数中 发一个邮件 通知 相关 的人员，手动处理异常，检查问题等。

retry_error_callback 这里可以传入一个回调 函数 ， 回函函数中有一个参数 retry_state 最后一次 函数的返回的状态。

在回调函数中 可以做你想做的任何处理，发邮件 或者其他的操作。

from tenacity import stop_after_attempt, retry, retry_if_result def send_email(msg): print(msg) # 回调函数 def callback(retry_state): """return the result of the last call attempt""" # print(f"call function ... {retry_state}, {type(retry_state)}") send_email('eventually_return_false eventually failed') return retry_state.outcome.result() def is_false(value): """Return True if value is False""" return value is False # will return False after trying 3 times to get a different result @retry(stop=stop_after_attempt(3), retry_error_callback=callback, retry=retry_if_result(is_false)) def eventually_return_false(): print("failed ...") return False if __name__ == '__main__': eventually_return_false() pass

7 错误处理

有的时候 我们可能重试后失败了，想获取原来 程序代码中抛出 的异常。 我们可以使用 reraise 参数来 抛出异常。是一个bool 变量，

设置为True 会抛出原来的异常， 如果设置为False 则不会抛出原来的异常 ,会抛出 tenacity.RetryError

from tenacity import retry, stop_after_delay, stop_after_attempt, retry_if_exception_type, retry_if_result class NotExistResourceErr(Exception): """ 定义一个自定义的异常 """ code = str(400) detail = "Parameter Error" def __init__(self, code: str = None, detail: Any = None): self.code = code or self.code self.detail = detail or self.detail @retry(reraise=True , stop=stop_after_attempt(3)) def raise_my_exception(): raise NotExistResourceErr(detail="Fail",code='0000') try: raise_my_exception() except NotExistResourceErr as e : print(f'handle exception detail:{e.detail} code:{e.code}')

设置为 reraise=False 整个程序 就会挂掉，如下面的报错：

Traceback (most recent call last):
  File "/Users/frank/code/venv38/venv/lib/python3.8/site-packages/tenacity/__init__.py", line 407, in __call__
    result = fn(*args, **kwargs)
  File "/Users/frank/code/py_proj/study-fastapi/my_retry_demo/hello6.py", line 42, in raise_my_exception
    raise NotExistResourceErr("Fail")
__main__.NotExistResourceErr: Fail

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/frank/code/py_proj/study-fastapi/my_retry_demo/hello6.py", line 45, in
    raise_my_exception()
  File "/Users/frank/code/venv38/venv/lib/python3.8/site-packages/tenacity/__init__.py", line 324, in wrapped_f
    return self(f, *args, **kw)
  File "/Users/frank/code/venv38/venv/lib/python3.8/site-packages/tenacity/__init__.py", line 404, in __call__
    do = self.iter(retry_state=retry_state)
  File "/Users/frank/code/venv38/venv/lib/python3.8/site-packages/tenacity/__init__.py", line 361, in iter
    raise retry_exc from fut.exception()
tenacity.RetryError: RetryError[]

8 推荐看下 作者的对这个库介绍

做了一些翻译，感兴趣的同学 可以关注一下。Python never gives up: the tenacity library 原文

Python never gives up: the tenacity library

A couple of years ago, I wrote about the Python retrying library. This library was designed to retry the execution of a task when a failure occurred.

几年前，我写过关于Python重试库的文章。这个库被设计用来在一个任务发生故障时重试执行。

I started to spread usage of this library in various projects, such as Gnocchi, these last years. Unfortunately, it started to get very hard t

上一篇：Python制作数据分析透视表的方法详解_python_

下一篇：PyTorch中torch.utils.data.DataLoader实例详解_python_

相关内容

点击排行

本栏推荐

猜你喜欢

-六神源码网