本篇文章介绍airbus中任务重启装饰器,用于当任务失败时,会根据指定次数和间隔时长来重新运行函数。

1. 源码

  1. import time
  2. import logging
  3. from functools import wraps, partial
  4. def retry(func = None, *, retry_times: int = 5, sleep_time: int = 60):
  5. log = logging.getLogger("Retry")
  6. if func is None:
  7. return partial(retry, retry_times=retry_times, sleep_time=sleep_time)
  8. @wraps(func)
  9. def wrapper(*args, **kwargs):
  10. cnt = 1
  11. while cnt <= retry_times:
  12. try:
  13. res = func(*args, **kwargs)
  14. return res
  15. except Exception as e:
  16. log.error(f"Retry [red]{cnt}[/] times \n", extra={"markup": True})
  17. cnt += 1
  18. time.sleep(sleep_time)
  19. log.error("Messages: [red]{0}[/]".format(e), extra={"markup": True})
  20. else:
  21. log.exception(f"Retry {retry_times} times, but all failed!")
  22. raise Exception(f"Retry {retry_times} times, but all failed!")
  23. return wrapper

2. 使用方法

  1. import time
  2. from airbus.decorators import retry
  3. @retry
  4. def retry_func():
  5. print(1 / 0)
  6. @retry(retry_times=2)
  7. def retry_func():
  8. print(1 / 0)
  9. @retry(retry_times=2, sleep_time=3)
  10. def retry_func():
  11. print(1 / 0)