前言
1、在实际应用中,经常会碰到在web网络请求时,因为网络的不稳定,会有请求超时的问题,这时候,一般都是自己去实现重试请求的逻辑,直到得到响应或者超时。虽然这样的逻辑并不复杂,但是代码写起来却不那么优雅,不那么pythonic。
2、在与接口的通信过程中,为了防止由于网络不稳定情况,造成请求错误或者超时等问题,或者其他不可控因素等造成功能性问题,代码中一般都会加入重试功能以增加代码的健壮性。
3、 tenacity
4、 tenacity
5、 tenacity
- 通用装饰器 API
- 指定停止条件(即限制尝试次数)
[En]
specify stop condition (that is, limit the number of attempts)*
- 指定等待条件(即两次尝试之间的指数退避睡眠)
[En]
specify waiting conditions (that is, exponential Backoff sleep between attempts)*
- 自定义重试异常
- 自定义预期结果重试
[En]
customize the retry of the expected result*
- 重试协程
- 使用上下文管理器重试代码块
[En]
retry the code block using the context manager*
6、在编写程序尤其是与网络请求相关的程序,如调用web接口、运行网络爬虫等任务时,经常会遇到一些偶然发生的请求失败的状况,这种时候如果我们仅仅简单的捕捉错误然后跳过对应任务,肯定是不严谨的,尤其是在网络爬虫中,会存在损失有价值数据的风险。
安装
pip install tenacity
使用
1、基本重试(无条件重试,重试之间无间隔,报错之后就立马重试)
tenacity 库的错误重试核心功能由其 retry 装饰器来实现,默认在不给 retry
from tenacity import retry@retrydef never_give_up_never_surrender(): print("无条件重试,重试之间无间隔,报错之后就立马重试") raise Exceptionif __name__ == '__main__': never_give_up_never_surrender()
运行结果:
2、停止重试
设置重试次数:
有些时候我们对某段函数逻辑错误重试的忍耐是有限度的,譬如当我们调用某个网络接口时,如果连续n次都执行失败,我们可能就会认为这个任务本身就存在缺陷,不是通过重试就能够使其正常的。
这种时候我们可以利用 tenacity 库中的 stop_after_attempt 函数,作为 retry() 中的 stop 参数传入,从而为我们”无尽”的错误重试过程添加一个终点,其中 stop_after_attempt()
from tenacity import retry, stop_after_attempt@retry(stop=stop_after_attempt(max_attempt_number=7))def stop_after_7_attempts(): print("重试7次后停止") raise Exceptionif __name__ == '__main__': stop_after_7_attempts()
运行结果:
设置时间限制:
tenacity 库还为我们提供了 stop_after_delay()
from tenacity import retry, stop_after_delay@retry(stop=stop_after_delay(10))def stop_after_10_s(): print("10秒后停止重试") raise Exceptionif __name__ == '__main__': stop_after_10_s()
运行结果:
组合多个停止条件:
如果我们的任务同时需要添加 最大重试次数以及 最大超时时长限制,在 tenacity 库中仅需要用|运算符组合不同的限制条件再传入 retry() 的 stop
将 stop_after_delay 函数和 stop_after_attempt
例如,当我们的函数执行重试超过3秒或超过5次时,我们可以结束重试:
[En]
For example, when our function executes a retry for more than 3 seconds or more than 5 times, we can end the retry:
from tenacity import retry, stop_after_delay, stop_after_attempt@retry(stop=(stop_after_delay(max_delay=3) | stop_after_attempt(max_attempt_number=5)))def stop_after_3_s_or_5_retries(): print("10秒后或者5次重试后停止重试") raise Exceptionif __name__ == '__main__': stop_after_3_s_or_5_retries()
运行结果:
如您所见,在下面的结果中,达到了最大重试5次的限制,从而结束了函数执行的重试过程。
[En]
As you can see, in the following results, the limit of “maximum retry 5 times” is reached, thus ending the retry process of function execution.
3、重试前等待(设置相邻函数重试之间的时间间隔)
有些情况下我们并不希望每一次重试抛出错误后,立即开始下一次的重试,譬如爬虫任务中为了更好地伪装我们的程序, tenacity 库中提供了一系列非常实用的函数,配合 retry() 装饰器中的 wait
重试之前等待固定时间:
通过使用 tenacity 库中的 wait_fixed()
from tenacity import retry, wait_fixed@retry(wait=wait_fixed(2))def wait_2_s(): print("每次重试前等待2秒") raise Exceptionif __name__ == '__main__': wait_2_s()
运行结果:
等待随机时间:
除了设置固定的时间间隔外, tenacity 库还可以通过 wait_random()
from tenacity import retry, wait_random@retry(wait=wait_random(min=1, max=2))def wait_random_1_to_2_s(): print("每次重试前等待1-2秒之间") raise Exceptionif __name__ == '__main__': wait_random_1_to_2_s()
运行结果:
等待指数时间:
from tenacity import retry, wait_exponential@retry(wait=wait_exponential(multiplier=1, min=4, max=10))def wait_exponential_1(): print("每次重试等待 2^x * multiplier 秒,x为重试次数,最小4秒,最多10秒") raise Exceptionif __name__ == '__main__': wait_exponential_1()
运行结果:
结合固定和抖动等待:
from tenacity import retry, wait_random, wait_fixed@retry(wait=wait_fixed(3) + wait_random(0, 2))def wait_fixed_jitter(): print("每次重试前等待3到3+2秒之间") raise Exceptionif __name__ == '__main__': wait_fixed_jitter()
运行结果:
链式等待:
from tenacity import retry, wait_fixed, wait_chain@retry(wait=wait_chain(*[wait_fixed(3) for i in range(3)] + [wait_fixed(7) for i in range(2)] + [wait_fixed(9)]))def wait_fixed_chained(): print("前3次等待3秒,接下来2次等待7秒,后面等待9秒") raise Exceptionif __name__ == '__main__': wait_fixed_chained()
运行结果:
4、自定义是否触发重试机制
tenacity 库中 retry()
捕捉或忽略特定的错误类型:
当函数中抛出的异常类型与 @retry() 装饰器中定义的异常类型相同时则执行重试策略,否则函数执行一次时直接抛出异常。(可以通过 retry_if_exception_type
from tenacity import retry, retry_if_exception_type, retry_if_not_exception_type@retry(retry=retry_if_exception_type(FileExistsError))def demo_func7(): raise TimeoutError@retry(retry=retry_if_not_exception_type(FileNotFoundError))def demo_func8(): raise FileNotFoundErrorif __name__ == '__main__': demo_func7() # demo_func8()
运行结果:
自定义函数结果条件判断函数
通过可以编写额外的条件判断函数,配合 tenacity 库中的 retry_if_result()
import randomfrom tenacity import retry, retry_if_result@retry(retry=retry_if_result(lambda x: x >= 0.1))def demo_func9(): a = random.random() print(a) return a# 记录开始时间
运行结果:
5、对函数的错误重试情况进行统计
被 tenacity 库中的 retry() 装饰的函数,可以通过打印其 retry.statistics
import randomfrom tenacity import retry, retry_if_result@retry(retry=retry_if_result(lambda x: x >= 0.1))def demo_func9(): a = random.random() print(a) return a# 记录开始时间demo_func9()print(demo_func9.retry.statistics)
运行结果:
6、重试错误后的异常抛出
出现异常后,函数会进行重试,若重试后还是失败,默认情况下,程序最后抛出的异常会变成 RetryError
因此可以在 @retry() 装饰器中加一个参数 reraise=True
1)不加 reraise=True
from tenacity import retry, stop_after_attempt@retry(stop=stop_after_attempt(3))def task(): print("task running ... ") raise Exceptiontask()
运行结果:
2)加 reraise=True
from tenacity import retry, stop_after_attempt@retry(stop=stop_after_attempt(3), reraise=True)def task(): print("task running ... ") raise Exceptiontask()
运行结果:
7、在函数重试前执行动作
tenacity
在函数执行前添加日志,然后执行函数,抛出异常后重新执行函数,然后在重新执行函数之前添加日志,依此类推。不,不,不。
[En]
Add the log before the function executes, then execute the function, then re-execute the function after throwing an exception, then add the log before re-executing the function, and so on. no, no, no.
from tenacity import retry, stop_after_attempt, before_logimport loggingimport sys logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)logger = logging.getLogger(__name__) @retry(stop=stop_after_attempt(3), before=before_log(logger=logger, log_level=logging.DEBUG))def task(): print("task running ... ") raise Exception task()
运行结果:
8、在函数重试后执行操作
与函数重试之前的操作类似:
[En]
Similar to what you did before the function retried:
import timefrom tenacity import retry, stop_after_attempt, after_logimport loggingimport syslogging.basicConfig(stream=sys.stderr, level=logging.DEBUG)logger = logging.getLogger(__name__)@retry(stop=stop_after_attempt(3), after=after_log(logger=logger, log_level=logging.DEBUG))def task(): print("task running ... ") time.sleep(2) raise Exceptiontask()
运行结果:
9、自定义回调函数(当最后一次重试失败后,可以执行一个回调函数)
定义函数最后一次重试仍然失败调用的函数(只在最后一次重试失败时调用),回调函数应该接受一个被调用的参数 retry_state
@retry 装饰器的定义回调函数的参数为: retry_error_callback=函数名
示例1:
from tenacity import stop_after_attempt, retry, retry_if_resultdef return_last_value(retry_state): """return the result of the last call attempt""" print('执行回调函数') print(retry_state.outcome.result()) 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=return_last_value, retry=retry_if_result(is_false))def eventually_return_false(): print('函数重试') return Falseif __name__ == '__main__': eventually_return_false()
运行结果:
retry_state 参数是 RetryCallState
- start_time(float)
- retry_object(BaseRetrying)
- fn(callable)
- args(tuple)
- kwargs(dict)
- attempt_number(int)
- outcome(tenacity.FutureorNone)
- outcome_timestamp(floatorNone)
- idle_for(float)
- next_action(tenacity.RetryActionorNone)
示例2:
from tenacity import retry, stop_after_attempt, retry_if_resultdef return_last_value(retry_state): print("执行回调函数") return retry_state.outcome.result() # 表示原函数的返回值def is_false(value): print('执行《判断是否进行被装饰函数重试》的函数') return value is False@retry(stop=stop_after_attempt(3), retry_error_callback=return_last_value, retry=retry_if_result(is_false))def test_retry(): print("等待重试.....") return Falseprint(test_retry())
运行结果:
https://zhuanlan.zhihu.com/p/391812968
期待陌生人,拥抱惊喜。
[En]
To expect strangers, to embrace surprises.
Original: https://blog.51cto.com/u_15688254/5508199
Author: mb62abf3afb54fb
Title: python之第三方库tenacity重试库的详细使用:Tenacity是一个通用的retry库,简化为任何任务加入重试的功能
原创文章受到原创版权保护。转载请注明出处:https://www.johngo689.com/513255/
转载文章受原作者版权保护。转载请注明原作者出处!