当前位置: 首页 > 科技观察

鲜为人知的Python“重试机制”

时间:2023-03-21 10:19:10 科技观察

旨在避免因某些网络或其他不可控因素导致的功能问题。比如发送请求的时候,因为网络不稳定,经常会出现请求超时的问题。这种情况下,我们通常会在代码中加入重试代码。重试代码本身并不难实现,但是如何写得优雅好用才是我们需要考虑的。这里要介绍的是一个第三方库——Tenacity(题目中的重试机制并不准确,它不是Python的内置模块,所以不能称之为机制),它实现了我们几乎所有的东西可以使用所有重试场景,比如:什么情况下重试?你重试了多少次?结束重试需要多长时间?每次重试间隔多长时间?首先安装它$pipinstalltenacity1。最基本的重试无条件重试,无间隔重试fromtenacityimportretry@retrydeftest_retry():print("等待重试,无间隔执行重试...")raiseExceptiontest_retry()无条件重试try,但等待2秒后再重试fromtenacityimportretry,wait_fixed@retry(wait=wait_fixed(2))deftest_retry():print("Waitingforretry...")raiseExceptiontest_retry()2.设置停止基本条件只重试7次fromtenacityimportretry,stop_after_attempt@retry(stop=stop_after_attempt(7))deftest_retry():print("waitingforretry...")raiseExceptiontest_retry()10秒后重试不重试fromtenacityimportretry,stop_after_delay@retry(stop=stop_after_delay(10))deftest_retry():print("waitingforretry...")raiseExceptiontest_retry()或者满足以上两个条件之一,重试结束fromtenacityimportretry,stop_after_delay,stop_after_attempt@retry(stop=(stop_after_delay(10)|stop_after_attempt(7)))deftest_retry():print(“哇正在重试...")raiseExceptiontest_retry()3。设置何时重试在特定错误/异常(比如请求超时)的情况下,重试fromrequestsimportexceptionsfromtenacityimportretry,retry_if_exception_type@retry(retry=retry_if_exception_type(exceptions.Timeout))deftest_retry():print("Waitingforretry...")raiseexceptions.Timeouttest_retry()在满足用户定义的条件时重试。下面的例子,当test_retry函数的返回值为False时,重试fromtenacityimportretry,stop_after_attempt,retry_if_resultdefis_false(value):returnvalueisFalse@retry(stop=stop_after_attempt(3),retry=retry_if_result(is_false))deftest_retry():returnFalsetest_retry()4.重试后重新抛出错误当出现异常时,韧度会重试。它仍然失败。默认情况下,上面抛出的异常会变成RetryError,而不是rootcause。所以可以加一个参数(reraise=True),这样当重试失败时,抛出的异常还是原来的异常。fromtenacityimportretry,stop_after_attempt@retry(stop=stop_after_attempt(7),reraise=True)deftest_retry():print("waitingforretry...")raiseExceptiontest_retry()5。设置上次重试失败后执行的回调函数一个回调函数fromtenacityimport*defreturn_last_value(retry_state):print("执行回调函数")returnretry_state.outcome.result()#表示返回原函数的返回值defis_false(value):returnvalueisFalse@retry(stop=stop_after_attempt(3),retry_error_callback=return_last_value,retry=retry_if_result(is_false))deftest_retry():print("Waitingforretry...")returnFalseprint(test_retry())输出如下等待重试...等待重试...等待重试Trial...执行回调函数False