我们在写程序的时候,尤其是网络请求相关的程序,比如调用web接口,运行网络爬虫等任务,我们经常会遇到一些偶然的请求失败,这个时候,如果我们简单地捕获错误然后跳过相应的任务,这肯定是不精确的,尤其是在网络爬虫中,会有丢失有价值数据的风险。在这种情况下,我们非常有必要在我们的程序逻辑中加入一些“错误重试”的策略。费老师,几年前我写过一篇介绍Python中tenacity库的文章,但是功能单一,只能满足基本需求。今天要给大家介绍的tenacity库,可能是目前Python生态中最好的错误重试库。下面就来看看它的主要功能吧!tenacity中的常用函数作为第三方Python库,我们可以使用pipinstalltenacity来安装。安装完成后,我们来学习一下tenacity的主要使用方法和特点:1.tenacity的基础知识。测试的核心功能是通过其重试装饰器实现的。当默认不给重试装饰器传递参数时,当它装饰的函数在运行过程中抛出错误时,它会一直重试。例如下面这个简单的例子:importrandomfromtenacityimportretry@retrydefdemo_func1():a=random.random()print(a)ifa>=0.1:raiseExceptiondemo_func1()如你所见,我们的函数体生成了一个每次在0到1之间的随机数,当随机数不超过0.1时,将停止抛错,否则每次抛错行为都会被韧度捕获并立即重试。2.设置最大重试次数有时候我们对于重试某个功能逻辑错误的容忍度是有限的。比如我们调用某个网络接口,如果连续n次执行失败,我们可能会认为这个任务本身存在缺陷,哪天重试就不正常了。这时候,我们可以使用tenacity中的stop_after_attempt函数作为retry()中的停止参数,为我们的“无休止”错误重试过程添加一个终点,其中stop_after_attempt()接受一个整数输入作为“最大重试次数”“Trytimes":fromtenacityimportretry,stop_after_attempt@retry(stop=stop_after_attempt(3))defdemo_func2():print('functionexecution')raiseExceptiondemo_func2()可以看出,我们的函数限制了最大权数之后oftrials,经过3次重试,第4次继续执行后仍然报错,正式抛出函数中对应的Exception错误,结束重试过程。3.设置重试的最大超时时间。tenacity除了像上一节一样设置最大错误重试次数,还为我们提供了stop_after_delay()函数来设置整个重试过程的最大耗时。它还将结束重试过程:importtimefromtenacityimportretry,stop_after_delay#设置最大重试超时为5秒@retry(stop=stop_after_delay(5))defdemo_func3():time.sleep(1)print(f'already过去{time.time()-start_time}秒')引发异常#记录开始时间start_time=time.time()demo_func3()4。结合重试停止条件如果我们的任务需要同时加上最大重试次数和最大超时时间限制。在坚韧中,你只需要使用|运算符组合不同的限制,然后传入retry()的停止参数。比如下面的例子,当我们的函数重试超过3秒或者次数大于5时都可以结束重试:importtimeimportrandomfromtenacityimportretry,stop_after_delay,stop_after_attempt@retry(stop=(stop_after_delay(3)|stop_after_attempt(5)))defdemo_func4():time.sleep(random.random())print(f'{time.time()-start_time}secondshavepassed')raiseException#记录开始时间start_time=time.time()demo_func4()您可以看到,在上面的演示中,“最大重试5次”达到了第一次限制,从而结束了重试过程。5.设置相邻重试的时间间隔在某些情况下,我们不希望每次重试抛出错误后立即开始下一次重试。比如为了在爬虫任务中更好的伪装我们的程序,tenacity中提供了一系列非常实用的函数,配合retry()的wait参数,帮助我们妥善处理相邻重试的时间间隔。比较实用的方法如下:(1)设置固定的时间间隔我们可以在tenacity中使用wait_fixed()来设置相邻重试之间的固定等待间隔,单位秒,就像下面这个简单的例子:importtimefromtenacityimportretry,wait_fixed,stop_after_attempt#设置重试等待间隔为1秒@retry(wait=wait_fixed(1),stop=stop_after_attempt(3))defdemo_func5():print(f'{time.time()-start_time}secondshaspassed')raiseException#记录开始时间start_time=time.time()demo_func5()(2)设置随机时间间隔除了设置固定的时间间隔,韧度还可以通过wait_random()帮我们设置相邻重试的均匀分布的随机数,只需要设置均匀分布的范围就可以:importtimefromtenacityimportretry,wait_random,stop_after_attempt#设置重试等待间隔为1到3之间的随机数@retry(wait=wait_random(min=1,max=3),stop=stop_after_attempt(5))defdemo_func6():打印(f'{time.time()-start_time}secondshaspassed')raiseException#记录开始时间start_time=time.time()demo_func6()可以观察到,每次重试后等待的时间是随机的~6。自定义是否触发重试tenacity中retry()的默认策略是当它装饰的函数执行过程“throwsanyerror”时重试,但在某些情况下我我们需要的可能是特定错误类型的捕获/忽略,或者异常计算结果的捕获Tenacity也内置了相关的实用功能:(1)捕捉或忽略特定的错误类型在tenacity中使用retry_if_exception_type()和retry_if_not_exception_type(),配合retry()的重试参数,我们可以捕捉或忽略特定的错误类型忽略:fromtenacityimportretry,retry_if_exception_type,retry_if_not_exception_type@retry(retry=retry_if_exception_type(FileExistsError))defdemo_func7():raiseTimeoutError@retry(retry=retry_if_not_exception_type(FileNotFoundError))defdemo_func8():raiseFileNotFoundError(2)自定义义函数结果条件判断函数我们可以多写一个条件判断函数,配合tenacity中的retry_if_result()实现自定义条件判断函数的返回结果,只有返回True才会触发重试操作:importrandomfromtenacityimportretry,retry_if_result@retry(retry=retry_if_result(lambdax:x>=0.1))defdemo_func9():a=random.random()print(a)返回a#记录开始时间demo_func9()7.查看函数的错误重试统计tenacity的retry()修饰的函数,我们可以打印其retry.statistics属性,查看错误重试统计记录结果。比如这里我们打印前面执行的示例函数demo_func9()的统计结果:demo_func9.retry.statistics除了上面的函数,tenacity还有很多特殊的功能,可以和logging模块,异步函数,coroutines结合使用和其他Python函数来实现更高级的功能。有兴趣的朋友可以去https://githubub.com/jd/tenacity了解更多
