Python错误重试利器:tenacity库核心功能详解
本文深入解析Python中强大的错误重试库tenacity。通过retry装饰器,实现自动重试逻辑。详解如何配置最大重试次数(stop_after_attempt)、超时限制(stop_after_delay)及组合条件。介绍wait_fixed和wait_random设置重试间隔,以及retry_if_exception_type和retry_if_result自定义重试触发条件,助力构建高可用程序。
1 简介
在开发程序,特别是涉及网络交互的场景(如调用Web接口、运行网络爬虫)时,偶尔出现的请求失败是常见现象。若仅简单捕获错误并跳过任务,这种做法并不严谨,尤其在网络爬虫中,可能导致有价值数据的丢失。
因此,为程序逻辑添加「错误重试」策略显得尤为必要。此前曾介绍过Python中的retry库,但其功能较为单一,仅能满足基本需求。
本文将介绍tenacity库,这或许是目前Python生态中最为强大的错误重试库。接下来,我们将详细探讨其核心功能。
2 tenacity中的常用功能
作为第三方Python库,可通过pip install tenacity进行安装。安装完成后,我们将学习tenacity的主要使用方法和特性:
2.1 tenacity的基础使用
tenacity的核心重试功能由retry装饰器实现。默认情况下,若不向retry装饰器传递参数,当被装饰的函数抛出错误时,它将无限次重试。以下是一个简单示例:
import random
from tenacity import retry
@retry
def demo_func1():
a = random.random()
print(a)
if a >= 0.1:
raise Exception
demo_func1()

如示例所示,函数内部每次生成0到1之间的随机数。当随机数不超过0.1时,停止抛出错误;否则,tenacity会捕捉每次错误并立即重试。
2.2 设置最大重试次数
有时,我们对函数逻辑错误重试的容忍度是有限的。例如,调用某个网络接口时,如果连续n次执行失败,可能意味着任务本身存在缺陷,重试无法解决问题。
此时,可利用tenacity中的stop_after_attempt函数,作为retry()的stop参数传入,从而为“无尽”的重试过程设定终点。stop_after_attempt()接受一个整数,表示「最大重试」次数:
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
def demo_func2():
print('函数执行')
raise Exception
demo_func2()

如示例所示,限制最大重试次数后,函数经过3次重试,在第4次继续执行并抛出错误后,正式抛出Exception错误并结束重试过程。
2.3 设置重试最大超时时长
除了设置最大重试次数外,tenacity还提供了stop_after_delay()函数,用于设置整个重试过程的最大耗时。超出该时长,重试过程也将结束:
import time
from tenacity import retry, stop_after_delay
# 设置重试最大超时时长为5秒
@retry(stop=stop_after_delay(5))
def demo_func3():
time.sleep(1)
print(f'已过去 {time.time() - start_time} 秒')
raise Exception
# 记录开始时间
start_time = time.time()
demo_func3()

2.4 组合重试停止条件
若任务需同时限制最大重试次数和最大超时时长,在tenacity中仅需用|运算符组合不同限制条件,再传入retry()的stop参数即可。例如,当函数执行重试超过3秒或次数大于5次时,均可结束重试:
import time
import random
from tenacity import retry, stop_after_delay, stop_after_attempt
@retry(stop=(stop_after_delay(3) | stop_after_attempt(5)))
def demo_func4():
time.sleep(random.random())
print(f'已过去 {time.time() - start_time} 秒')
raise Exception
# 记录开始时间
start_time = time.time()
demo_func4()

如演示所示,先达到“最大重试5次”的限制,从而结束了重试过程。
2.5 设置相邻重试之间的时间间隔
有时,我们并不希望每次重试抛出错误后立即开始下一次重试。例如,在爬虫任务中,为了更好地伪装程序,tenacity提供了一系列实用函数,配合retry()的wait参数,妥善处理相邻重试之间的时间间隔。主要实用方式有两种:
2.5.1 设置固定时间间隔
通过tenacity中的wait_fixed(),可为相邻重试设置固定的等待间隔秒数,如下简单示例:
import time
from tenacity import retry, wait_fixed, stop_after_attempt
# 设置重试等待间隔为1秒
@retry(wait=wait_fixed(1), stop=stop_after_attempt(3))
def demo_func5():
print(f'已过去 {time.time() - start_time} 秒')
raise Exception
# 记录开始时间
start_time = time.time()
demo_func5()

2.5.2 设置随机时间间隔
除固定间隔外,tenacity还可通过wait_random()为相邻重试设置均匀分布的随机数,只需设置好均匀分布的范围即可:
import time
from tenacity import retry, wait_random, stop_after_attempt
# 设置重试等待间隔为1到3之间的随机数
@retry(wait=wait_random(min=1, max=3), stop=stop_after_attempt(5))
def demo_func6():
print(f'已过去 {time.time() - start_time} 秒')
raise Exception
# 记录开始时间
start_time = time.time()
demo_func6()

如观察所示,每次重试后的等待时长均为随机值。
2.6 自定义是否触发重试
tenacity中retry()的默认策略是:当被装饰的函数执行过程“抛出任何错误”时即进行重试。但有时,我们需要对特定错误类型进行捕捉/忽略,或对异常计算结果进行捕捉。
tenacity中同样内置了相关实用功能:
2.6.1 捕捉或忽略特定的错误类型
使用tenacity中的retry_if_exception_type()和retry_if_not_exception_type(),配合retry()的retry参数,可对特定错误类型进行捕捉或忽略:
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 FileNotFoundError


2.6.2 自定义函数结果条件判断函数
我们可以编写额外的条件判断函数,配合tenacity中的retry_if_result(),实现对函数返回结果的自定义条件判断。仅当返回True时,才会触发重试操作:
import random
from 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()

2.7 对函数的错误重试情况进行统计
被tenacity的retry()装饰的函数,可通过打印其retry.statistics属性查看历经的错误重试统计记录。例如,对前面执行过的示例函数demo_func9()的统计结果进行打印:
demo_func9.retry.statistics

除上述功能外,tenacity还具有许多特殊特性,可结合logging模块、异步函数、协程等其他Python功能实现更高级的功能。感兴趣的朋友可前往https://github.com/jd/tenacity了解更多。
到此这篇关于Python中最强大的错误重试库(tenacity库)的文章就介绍到这了,更多相关Python tenacity错误重试库内容请搜索本站以前的文章或继续浏览下面的相关文章希望大家以后多多支持本站!
本站声明
本文内容仅供参考,不代表本站观点。转载或引用请注明出处。如发现内容问题,请通过联系我们页面反馈。













