- 1、 引言
- 2、 tenacity
- 2.1 安装
- 2.2 基本用法
- 2.3 重试最大次数
- 2.4 重试最长时间
- 2.5 组合重试停止条件
- 2.6 相邻重试时间间隔
- 2.6.1 固定时间间隔
- 2.6.2 随机时间间隔
- 2.7 自定义是否触发重试
- 2.7.1 忽略特定错误类型
- 2.7.2 自定义函数结果条件判断函数
- 3、总结
小屌丝:鱼哥,最近遇到点问题。
小鱼:呦呵,你这是遇到的问题不小啊,不然不能主动跟我说。
小屌丝:唉~~ 在试错的路上,越走越迷茫。
小鱼:那是因为你没有用对方法啊。
小屌丝:这个还有方法?
小鱼:这你就不懂了吧。
小屌丝:那你快快教教我哦~ ~
小鱼:retry 你可以了解一下。
小屌丝:retry用法单一,不符合我的要(wei)求(kou)。
小鱼:那你就试一试tenacity
小屌丝:不会,你教我啊。
小鱼:… 这又是赔本的一天。
我们在调用web接口,或者爬虫的时候,经常会遇到一些偶发请求失败的状况,这个时候,我们可能会想到一直循环尝试直到成功;或者打出错误信息;
这种做法,在简单的程序中没有错,但是,在复杂的或者大型项目中,这种做法显然有很多弊端。
而我们需要用到一个错误重试方法,老鸟应该会想到是tenactiy。
没错,tenacity可能目前是python中最好的第三方重试库,
接下来,我们就来一睹它的风采。
作为第三方库,我们第一步必须是 安装
pip install tenacity
其他方式安装:
《Python3,选择Python自动安装第三方库,从此跟pip说拜拜!!》
《Python3:我低调的只用一行代码,就导入Python所有库!!》
tenacity的错误重试核心功能由其retry装饰器来实现,
默认不给retry装饰器传参数时,它会在其所装饰的函数运行过程抛出错误不时停地重试下去。
代码示例
# -*- coding:utf-8 -*-
# @Time : 2022-04-04
# @Author : carl_DJ
import random
from tenacity import retry
@retry
def demo_one():
a = random.random()
print(f'{a}')
if a >= 0.1:
raise Exception
demo_one()
运行结果
通过结果,可以看到,函数体内每次生成0到1之间的随机数,当这个随机数不超过0.1时才会停止抛出错误,否则则会被tenacity捕捉到每次的错误抛出行为并立即重试。
由于我们的时间是宝贵的,所以重试的次数也需要有限制的。
我们可以利用tenacity中的stop_after_attempt函数,作为retry()中的stop参数传入,从而为我们“无尽”的错误重试过程添加一个终点,
其中**stop_after_attempt()**接受一个整数输入作为"最大重试"的次数:
代码示例
# -*- coding:utf-8 -*-
# @Time : 2022-04-04
# @Author : carl_DJ
from tenacity import retry,stop_after_attempt
#设置错误重试,3次
@retry(stop = stop_after_attempt(3))
def demo_two():
print(f'函数执行')
raise Exception
demo_two()
运行结果
通过结果,可以看到,在第4次继续执行正式地抛出了函数中对应的Exception错误结束了重试过程。
除了设置错误的次数,还可以设置最大的重试耗时,
通过stop_after_delay() 的函数来设置,超过这个时长就会结束重试过程。
代码示例
# -*- coding:utf-8 -*-
# @Time : 2022-04-04
# @Author : carl_DJ
import time
from tenacity import retry,stop_after_delay
#设置重试最大超时时长为4秒
@retry(stop = stop_after_delay(4))
def demo_three():
#每次时间间隔为2秒
time.sleep(2)
print(f'已过去 {time.time() - start_time} 秒')
raise Exception
start_time = time.time()
demo_three()
运行结果
如果同时需要添加最大重试次数以及最大超时时长限制,
在tenacity中仅需要用 |
运算符组合不同的限制条件再传入retry()的stop参数即可,
代码示例
# -*- coding:utf-8 -*-
# @Time : 2022-04-04
# @Author : carl_DJ
import time
import random
from tenacity import retry,stop_after_attempt,stop_after_delay
#函数执行重试超过3秒或次数大于5次时均可以结束重试
@retry(stop =(stop_after_attempt(3) | stop_after_delay(5)))
def demo_four():
time.sleep(random.random())
print(f'已过去 {time.time() - start_time} 秒')
raise Exception
#开始时间
start_time = time.time()
demo_four()
运行结果
tenacity中提供了一系列非常实用的函数,配合retry()的wait参数,帮助我们妥善处理相邻重试之间的时间间隔,其中较为实用的主要有以下两种方式:
- 固定时间间隔;
- 随机时间间隔;
使用tenacity中的**wait_fixed()**可以为相邻重试之间设置固定的等待间隔秒数
代码示例
# -*- coding:utf-8 -*-
# @Time : 2022-04-04
# @Author : carl_DJ
import time
from tenacity import retry,wait_fixed,stop_after_attempt
# 设置重试等待间隔为1秒
@retry(wait = wait_fixed(1),stop = stop_after_attempt(3))
def demo_five():
print(f'已过去 {time.time() - start_time} 秒')
raise Exception
#开始时间
start_time = time.time()
demo_five()
运行结果
除了设置固定的时间间隔外,tenacity还可以通过**wait_random()**为相邻重试设置均匀分布随机数,只需要设置好均匀分布的范围即可:
代码示例
# -*- coding:utf-8 -*-
# @Time : 2022-04-04
# @Author : carl_DJ
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(4))
def demo_six():
print(f'已过去 {time.time() - start_time} 秒')
raise Exception
#开始时间
start_time = time.time()
demo_six()
运行结果
2.7 自定义是否触发重试tenacity中retry()的默认策略是当其所装饰的函数执行过程“抛出任何错误”时即进行重试,
但有些情况下需要的可能是对特定错误类型的捕捉/忽略,亦或是对异常计算结果的捕捉。
tenacity中同样内置了相关的实用功能:
- 捕捉或忽略特定的错误类型;
- 自定义函数结果条件判断函数;
使用tenacity中的retry_if_exception_type()和retry_if_not_exception_type(),配合retry()的retry参数,对特定的错误类型进行捕捉或忽略:
代码示例
# -*- coding:utf-8 -*-
# @Time : 2022-04-04
# @Author : carl_DJ
'''
捕捉或忽略特定的错误类型
'''
from tenacity import retry,retry_if_exception_type,retry_if_not_exception_type
#retry_if_exception_type()
@retry(retry=retry_if_exception_type(FileExistsError))
def demo_seven():
raise TimeoutError
demo_seven()
#retry_if_not_exception_type()
@retry(retry=retry_if_not_exception_type(FileNotFoundError))
def demo_eight():
raise FileNotFoundError
demo_eight()
运行结果
demo_seven()
demo_eight()
我们可以编写额外的条件判断函数,配合tenacity中的retry_if_result(),实现对函数的返回结果进行自定义条件判断,返回True时才会触发重试 *** 作:
代码示例
# -*- coding:utf-8 -*-
# @Time : 2022-04-04
# @Author : carl_DJ
import random
from tenacity import retry,retry_if_result
@retry(retry = retry_if_result(lambda x: x >= 0.1))
def demo_nine():
a = random.random()
print(f'{a}')
return a
demo_nine()
运行结果
今天的分享,就到这里了。
是不是奇奇怪怪的姿势,又增加了呢。
在错误的重试姿势,也增加了。
关注小鱼,学习更多关于python第三方库的知识。
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)