智能重试库
retrylib的Python项目详细描述
重试decorator参数
重试(尝试次数,延迟=0,步骤=0,最大延迟=1,重试打开=异常,记录器=无)
- 尝试次数:函数调用次数(首次调用+重试)。如果尝试次数<;0,则无限次重试
- 延迟:第一次重试前的延迟
- 步骤:每次重试时超时的增量值 <最大延迟值(延迟上限)<
- 重试:应处理的异常或检查的函数 是否应执行重试(默认:异常)
- logger:logger写入警告
返回修饰函数的结果
针对特定异常重试
from retrylib import retry
@retry(attempts_number=3, retry_on=(MyException,))
def function():
raise MyException()
使用自定义函数
from retrylib import retry
def is_my_exception(error):
return isinstance(error, MyException)
@retry(attempts_number=3, retry_on=is_my_exception)
def function():
raise MyException()
网络错误时重试
可以使用以下代码为自定义网络添加重试次数 功能:
import requests
from retrylib.network import retry
@retry()
def function():
response = requests.get('http://localhost:5002')
response.raise_for_status()
return response
function()
记录
全局记录器:您可以将特定的记录器传递给decorator
import logging
import logging.config
from retrylib.network import retry
LOGGING = {
'version': 1,
'formatters': {
'precise': {
'datefmt': '%Y-%m-%d,%H:%M:%S',
'format': '%(levelname)-7s %(asctime)15s '
'%(name)s:%(lineno)d %(message)s'
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'precise',
'stream': 'ext://sys.stderr'
},
},
'root': {
'level': 'INFO',
'handlers': ['console']
}
}
logging.config.dictConfig(LOGGING)
LOGGER = logging.getLogger(__name__)
@retry(logger=LOGGER)
def function():
response = requests.get('http://localhost:5002')
response.raise_for_status()
return response
特定于对象的记录器:要使用特定于对象的记录器,请定义方法“get_logger”
from retrylib import retry
class MyClass(object):
def __init__(self):
self._logger = logging.getLogger(__name__)
def get_logger(self):
return self._logger
@retry()
def my_method(self):
...
raise Exception
obj = MyClass()
obj.my_method()
# obj._logger will be used