本篇文章介绍airbus
中任务重启装饰器,用于当任务失败时,会根据指定次数和间隔时长来重新运行函数。
1. 源码
import time
import logging
from functools import wraps, partial
def retry(func = None, *, retry_times: int = 5, sleep_time: int = 60):
log = logging.getLogger("Retry")
if func is None:
return partial(retry, retry_times=retry_times, sleep_time=sleep_time)
@wraps(func)
def wrapper(*args, **kwargs):
cnt = 1
while cnt <= retry_times:
try:
res = func(*args, **kwargs)
return res
except Exception as e:
log.error(f"Retry [red]{cnt}[/] times \n", extra={"markup": True})
cnt += 1
time.sleep(sleep_time)
log.error("Messages: [red]{0}[/]".format(e), extra={"markup": True})
else:
log.exception(f"Retry {retry_times} times, but all failed!")
raise Exception(f"Retry {retry_times} times, but all failed!")
return wrapper
2. 使用方法
import time
from airbus.decorators import retry
@retry
def retry_func():
print(1 / 0)
@retry(retry_times=2)
def retry_func():
print(1 / 0)
@retry(retry_times=2, sleep_time=3)
def retry_func():
print(1 / 0)