toolbox/porter/tasks/base_task.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import asyncio
from collections import defaultdict
import logging
import traceback

logger = logging.getLogger("toolbox")

from toolbox.porter.common.params import Params


global_file_lock_dict = defaultdict(asyncio.Lock)


class BaseTask(Params):
    def __init__(self,
                 flag: str,
                 check_interval: int,
                 ):
        super().__init__()
        self.flag = flag
        self.check_interval = check_interval

    async def do_task(self):
        raise NotImplementedError

    async def start(self):
        while True:
            try:
                await self.do_task()
                logger.info(f"{self.flag}任务检测... 刷新间隔 {self.check_interval}s")
                await asyncio.sleep(self.check_interval)
            except Exception as error:
                logger.error(f"{self.flag}任务检测出错\nerror type: {type(error)}, error text: {error}, traceback: {traceback.format_exc()}")
                await asyncio.sleep(self.check_interval)
                continue


if __name__ == "__main__":
    pass