• 1. 源码
• 2. 使用方法

本篇文章介绍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)