Asynchronous MySQL reads and writes are forcefully cancelled.

[tufbel]

I encountered a rather peculiar issue where setting a smaller time interval for Apscheduler tasks resulted in the "Not connected" problem during asynchronous execution of MySQL writes.

I modified a convenient-to-use base class for Apscheduler myself.

@dataclass
class BaseTask(metaclass=SingletonMeta):
    alive_flag: bool = False
    job_config: Dict = None
    job: Job = None

    __scheduler: AsyncIOScheduler = None

    def __init__(self, *args, **kwargs):
        super().__init__()
        self.alive_flag = False
        self.job_config = kwargs
        self.job_config["id"] = self.__class__.__name__

    @classmethod
    def scheduler(cls) -> AsyncIOScheduler:
        if not isinstance(BaseTask.__scheduler, AsyncIOScheduler):
            BaseTask.__scheduler = AsyncIOScheduler()
        return cls.__scheduler

    @classmethod
    def start_scheduler(cls):
        logger.success(f"Start BaseForwardTask scheduler")
        cls.scheduler().start()

    @classmethod
    def shutdown_scheduler(cls, wait=True):
        logger.warning(f"Shutdown BaseForwardTask scheduler ...")
        cls.scheduler().shutdown(wait=wait)

    async def job_func(self):
        pass

    def start(self):
        if self.alive_flag:
            logger.info(f"{self} is started")
            return
        self.alive_flag = True
        self.job = self.scheduler().add_job(self.job_func, **self.job_config)
        logger.success(f"Start {self}")


    def stop(self):
        if not self.alive_flag:
            logger.info(f"{self} is stopped")
            return
        if isinstance(self.job, Job):
            self.job.remove()
        self.job = None
        logger.warning(f"{self} stop")
        self.alive_flag = False

Utilizing this base class, I have written a scheduled task.

class MyTask(BaseTask):
    redis_client = global_om["redis_client"]

    def __init__(self):
        super().__init__()
        self.job_config.update(
            {
                "trigger": CronTrigger(second="*/5"),
                "max_instances": 1,
            }
        )

    async def job_func(self):
        logger.debug("Running mytask")
        try:
            all_objs = await ObjClass.all()
        except Exception as exc:
            logger.error(f"Mytask error - {exc.__class__.__name__}:{exc}")  

        # ... Some additional logic code for handling reading and writing to Redis
        
        asyncio.create_task(self._delete_obj(all_objs))

    async def _delete_obj(self, all_objs):
        # ... Some additional logic code for handling reading and writing to Redis
        new_objs = await ObjClass.filter(id__in=[obj.id for obj_ in all_objs])
        # ...
        await ObjClass.filter(id__in=[obj.id for obj_ in all_objs]).delete()
        # ...

I am using the ORM framework Tortoise-ORM version 0.19.1, and I have defined an ObjClass. Currently, the task interval is set to 5 seconds. However, when I reduce the task interval to 2 seconds or any other value, after running the program for a while, I encounter a pymysql.InterfaceError(0, "Not connected"). This error consistently occurs at the line await ObjClass.filter(id__in=[obj.id for obj_ in all_objs]).delete(). The preceding query statement is always successful, and once this error occurs, it cannot be recovered from.