简单实现了一个在函数执行出现异常时自动重试的装饰器,支持控制最多重试次数,每次重试间隔,每次重试间隔时间递增。
最新的代码可以访问从github上获取
https://github.com/blackmatrix7/matrix-toolkit/blob/master/toolkit/retry.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
|
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2017/8/18 上午9:50
# @Author : Matrix
# @Github : https://github.com/blackmatrix7/
# @Blog : http://www.cnblogs.com/blackmatrix/
# @File : retry.py
# @Software: PyCharm
import time
from functools import wraps
__author__ = 'blackmatrix'
"""
在函数执行出现异常时自动重试的简单装饰器
"""
class StopRetry(Exception):
def __repr__( self ):
return 'retry stop'
def retry(max_retries: int = 5 , delay: ( int , float ) = 0 , step: ( int , float ) = 0 ,
exceptions: (BaseException, tuple , list ) = BaseException,
sleep = time.sleep, callback = None , validate = None ):
"""
函数执行出现异常时自动重试的简单装饰器。
:param max_retries: 最多重试次数。
:param delay: 每次重试的延迟,单位秒。
:param step: 每次重试后延迟递增,单位秒。
:param exceptions: 触发重试的异常类型,单个异常直接传入异常类型,多个异常以tuple或list传入。
:param sleep: 实现延迟的方法,默认为time.sleep。
在一些异步框架,如tornado中,使用time.sleep会导致阻塞,可以传入自定义的方法来实现延迟。
自定义方法函数签名应与time.sleep相同,接收一个参数,为延迟执行的时间。
:param callback: 回调函数,函数签名应接收一个参数,每次出现异常时,会将异常对象传入。
可用于记录异常日志,中断重试等。
如回调函数正常执行,并返回True,则表示告知重试装饰器异常已经处理,重试装饰器终止重试,并且不会抛出任何异常。
如回调函数正常执行,没有返回值或返回除True以外的结果,则继续重试。
如回调函数抛出异常,则终止重试,并将回调函数的异常抛出。
:param validate: 验证函数,用于验证执行结果,并确认是否继续重试。
函数签名应接收一个参数,每次被装饰的函数完成且未抛出任何异常时,调用验证函数,将执行的结果传入。
如验证函数正常执行,且返回False,则继续重试,即使被装饰的函数完成且未抛出任何异常。
如回调函数正常执行,没有返回值或返回除False以外的结果,则终止重试,并将函数执行结果返回。
如验证函数抛出异常,且异常属于被重试装饰器捕获的类型,则继续重试。
如验证函数抛出异常,且异常不属于被重试装饰器捕获的类型,则将验证函数的异常抛出。
:return: 被装饰函数的执行结果。
"""
def wrapper(func):
@wraps (func)
def _wrapper( * args, * * kwargs):
nonlocal delay, step, max_retries
func_ex = StopRetry
while max_retries > 0 :
try :
result = func( * args, * * kwargs)
# 验证函数返回False时,表示告知装饰器验证不通过,继续重试
if callable (validate) and validate(result) is False :
continue
else :
return result
except exceptions as ex:
# 回调函数返回True时,表示告知装饰器异常已经处理,终止重试
if callable (callback) and callback(ex) is True :
return
func_ex = ex
finally :
max_retries - = 1
if delay > 0 or step > 0 :
sleep(delay)
delay + = step
else :
raise func_ex
return _wrapper
return wrapper
if __name__ = = '__main__' :
pass
|
以上这篇python重试装饰器的简单实现方法就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持服务器之家。
原文链接:https://www.cnblogs.com/blackmatrix/p/7425435.html