一个计算机技术爱好者与学习者

0%

好好学Python:Python重试

1. 为什么需要重试?

典型场景:程序的实现需要调用第三方的API,但是我们并不能保证第三方API一直好用,也不能保证网络一直畅通,所以在调用第三方API时需要加上错误重试。

通用场景:程序的运行不符合预期,我们知道再次调用大概率可以使之符合预期,这时就需要重试。

而重试逻辑我们可以通过循环多次调用实现,也可以使用封装好的重试模块。
本文我们就学习一下Python中比较流行的 retrying 模块,实现一些重试Demo。

2. 安装retrying

pip install retrying

3. 任意异常重试

1
2
3
4
5
6
7
8
9
from retrying import retry

@retry(stop_max_attempt_number=3, wait_random_min=1000, wait_random_max=3000)
def run():
print('run')
raise ValueError

if __name__ == '__main__':
run()

当函数run抛出任意异常时,则进行重试。

  • stop_max_attempt_number:最大重试次数
  • wait_random_min:最小重试间隔
  • wait_random_max:最大重试间隔

注意函数内不要使用try except,否则异常就被函数自己捕获处理了,不会抛给retry。

4. 指定异常重试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from retrying import retry
import random

def if_value_error(exception):
return isinstance(exception, ValueError)

@retry(retry_on_exception=if_value_error, stop_max_attempt_number=3, wait_random_min=1000, wait_random_max=3000)
def run():
value = random.randint(0, 10)
msg = f'value is {value}'
print(msg)
if value <= 5:
raise ValueError('value is too small!')
else:
raise Exception('value is too large!')

if __name__ == '__main__':
run()

当函数run抛出ValueError异常时,则进行重试;抛出Exception异常时,不会重试。

5. 指定返回值重试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from retrying import retry
import random

def if_return_small(value):
return value <= 5

@retry(retry_on_result=if_return_small, stop_max_attempt_number=3, wait_random_min=1000, wait_random_max=3000)
def run():
value = random.randint(0, 10)
msg = f'value is {value}'
print(msg)
return value

if __name__ == '__main__':
run()

当函数run返回值小于5时,进行重试。

6. 多次重试后容忍异常

1
2
3
4
5
6
7
8
9
10
11
12
13
from retrying import retry

@retry(stop_max_attempt_number=3, wait_random_min=1000, wait_random_max=3000)
def run():
print('run')
raise ValueError

def run2():
print('run2')

if __name__ == '__main__':
run()
run2()

当函数run抛出任意异常时,则进行重试。重试指定次数后,程序就异常结束了。这时,如果我们想要容忍异常,让run2可以正常运行,该怎么办呢?很简单,给run函数包裹一个try except即可。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from retrying import retry

@retry(wrap_exception=True,stop_max_attempt_number=3, wait_random_min=1000, wait_random_max=3000)
def run():
print('run')
raise ValueError

def wrap_run():
try:
run()
except:
pass

def run2():
print('run2')

if __name__ == '__main__':
wrap_run()
run2()
  • 本文作者: 好好学习的郝
  • 原文链接: https://www.voidking.com/dev-python-retry/
  • 版权声明: 本文采用 BY-NC-SA 许可协议,转载请注明出处!源站会即时更新知识点并修正错误,欢迎访问~
  • 微信公众号同步更新,欢迎关注~