现在要求做一个重试逻辑,由于考虑到要做成docker镜像,减少依赖,于是自己写了一个异常重试的逻辑,代码如下:
import time
import logging
# logger模块
logger = logging.getLogger('fastone.cwl-runner')
logger.setLevel(level=logging.INFO)
formatter = logging.Formatter('%(asctime)s %(filename)s:%(lineno)d %(levelname)s %(message)s')
handler = logging.StreamHandler()
handler.setFormatter(formatter)
handler.setLevel(level=logging.INFO)
logger.addHandler(handler)
# 自定义异常
class RetryTimeoutError(Exception):
pass
# 装饰器模块,需要传入参数(重试次数)
def retry(retry_times):
def decorator(func):
def wrapper(*args, **kwargs):
retry_time = 0
while retry_time < retry_times:
try:
return func(*args, **kwargs)
except Exception as e:
retry_time += 1
time.sleep(3)
# 这里的exc_info=1 可以把Traceback打印到log里
logger.warning('ERROR: api_create has error: %s', e, exc_info=1)
else:
raise TypeError
return wrapper
return decorator
@retry(attempt=3)
def a():
a = 1
print(a)
raise RetryTimeoutError('重试超时')
if __name__ == '__main__':
a()
运行结构如下:

网友评论