Add scheduler to executor

Add scheduler to executor

Author: JE-Chen

je_load_density/utils/executor/action_executor.py

@@ -1,8 +1,8 @@
 import builtins
 import sys
-import time
 import types
 from inspect import getmembers, isbuiltin
+from typing import Union, Any
 
 from je_load_density.utils.exception.exception_tags import executor_data_error, add_command_exception_tag
 from je_load_density.utils.exception.exception_tags import executor_list_error
@@ -12,6 +12,7 @@
 from je_load_density.utils.generate_report.generate_xml_report import generate_xml, generate_xml_report
 from je_load_density.utils.json.json_file.json_file import read_action_json
 from je_load_density.utils.package_manager.package_manager_class import package_manager
+from je_load_density.utils.scheduler.extend_apscheduler import scheduler_manager
 from je_load_density.wrapper.start_wrapper.start_test import start_test
 
 
@@ -26,10 +27,17 @@ def __init__(self):
             "generate_json_report": generate_json_report,
             "generate_xml": generate_xml,
             "generate_xml_report": generate_xml_report,
-            # execute
+            # Execute
             "execute_action": self.execute_action,
             "execute_files": self.execute_files,
             "add_package_to_executor": package_manager.add_package_to_executor,
+            # Scheduler
+            "scheduler_event_trigger": self.scheduler_event_trigger,
+            "remove_blocking_scheduler_job": scheduler_manager.remove_blocking_job,
+            "remove_nonblocking_scheduler_job": scheduler_manager.remove_nonblocking_job,
+            "start_blocking_scheduler": scheduler_manager.start_block_scheduler,
+            "start_nonblocking_scheduler": scheduler_manager.start_nonblocking_scheduler,
+            "start_all_scheduler": scheduler_manager.start_all_scheduler,
         }
         # get all builtin function and add to event dict
         for function in getmembers(builtins, isbuiltin):
@@ -100,6 +108,16 @@ def execute_files(self, execute_files_list: list):
             execute_detail_list.append(self.execute_action(read_action_json(file)))
         return execute_detail_list
 
+    def scheduler_event_trigger(
+            self, function: str, id: str = None, args: Union[list, tuple] = None,
+            kwargs: dict = None, scheduler_type: str = "nonblocking", wait_type: str = "secondly",
+            wait_value: int = 1, **trigger_args: Any) -> None:
+        if scheduler_type == "nonblocking":
+            scheduler_event = scheduler_manager.nonblocking_scheduler_event_dict.get(wait_type)
+        else:
+            scheduler_event = scheduler_manager.blocking_scheduler_event_dict.get(wait_type)
+        scheduler_event(self.event_dict.get(function), id, args, kwargs, wait_value, **trigger_args)
+
 
 executor = Executor()
 package_manager.executor = executor

je_load_density/utils/scheduler/extend_apscheduler.py

@@ -1,5 +1,7 @@
-from typing import Callable
+from datetime import datetime
+from typing import Callable, Any, Union
 
+from apscheduler.job import Job
 from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.schedulers.blocking import BlockingScheduler
 from apscheduler.util import undefined
@@ -10,119 +12,204 @@ class SchedulerManager(object):
     def __init__(self):
         self._blocking_schedulers: BlockingScheduler = BlockingScheduler()
         self._background_schedulers: BackgroundScheduler = BackgroundScheduler()
+        self.blocking_scheduler_event_dict = {
+            "secondly": self.add_interval_blocking_secondly,
+            "minutely": self.add_interval_blocking_minutely,
+            "hourly": self.add_interval_blocking_hourly,
+            "daily": self.add_interval_blocking_daily,
+            "weekly": self.add_interval_blocking_weekly,
+        }
+        self.nonblocking_scheduler_event_dict = {
+            "secondly": self.add_interval_nonblocking_secondly,
+            "minutely": self.add_interval_nonblocking_minutely,
+            "hourly": self.add_interval_nonblocking_hourly,
+            "daily": self.add_interval_nonblocking_daily,
+            "weekly": self.add_interval_nonblocking_weekly,
+        }
 
     def add_blocking_job(
-            self, func, trigger=None, args=None, kwargs=None, id=None, name=None,
-            misfire_grace_time=undefined, coalesce=undefined, max_instances=undefined,
-            next_run_time=undefined, jobstore='default', executor='default',
-            replace_existing=False, **trigger_args):
+            self, func: Callable, trigger: str = None, args: Union[list, tuple] = None,
+            kwargs: dict = None, id: str = None, name: str = None,
+            misfire_grace_time: int = undefined, coalesce: bool = undefined, max_instances: int = undefined,
+            next_run_time: datetime = undefined, jobstore: str = 'default', executor: str = 'default',
+            replace_existing: bool = False, **trigger_args: Any) -> Job:
+        """
+        Just an apscheduler add job wrapper.
+        :param func: callable (or a textual reference to one) to run at the given time
+        :param str|apscheduler.triggers.base.BaseTrigger trigger: trigger that determines when
+            ``func`` is called
+        :param list|tuple args: list of positional arguments to call func with
+        :param dict kwargs: dict of keyword arguments to call func with
+        :param str|unicode id: explicit identifier for the job (for modifying it later)
+        :param str|unicode name: textual description of the job
+        :param int misfire_grace_time: seconds after the designated runtime that the job is still
+            allowed to be run (or ``None`` to allow the job to run no matter how late it is)
+        :param bool coalesce: run once instead of many times if the scheduler determines that the
+            job should be run more than once in succession
+        :param int max_instances: maximum number of concurrently running instances allowed for this
+            job
+        :param datetime next_run_time: when to first run the job, regardless of the trigger (pass
+            ``None`` to add the job as paused)
+        :param str|unicode jobstore: alias of the job store to store the job in
+        :param str|unicode executor: alias of the executor to run the job with
+        :param bool replace_existing: ``True`` to replace an existing job with the same ``id``
+            (but retain the number of runs from the existing one)
+        :return: Job
+        """
         params = locals()
         params.pop("self")
         trigger_args = params.pop("trigger_args")
-        self._blocking_schedulers.add_job(**params, **trigger_args)
+        return self._blocking_schedulers.add_job(**params, **trigger_args)
 
     def add_nonblocking_job(
-            self, func, trigger=None, args=None, kwargs=None, id=None, name=None,
-            misfire_grace_time=undefined, coalesce=undefined, max_instances=undefined,
-            next_run_time=undefined, jobstore='default', executor='default',
-            replace_existing=False, **trigger_args):
+            self, func: Callable, trigger: str = None, args: Union[list, tuple] = None,
+            kwargs: dict = None, id: str = None, name: str = None,
+            misfire_grace_time: int = undefined, coalesce: bool = undefined, max_instances: int = undefined,
+            next_run_time: datetime = undefined, jobstore: str = 'default', executor: str = 'default',
+            replace_existing: bool = False, **trigger_args: Any) -> Job:
+        """
+        Just an apscheduler add job wrapper.
+        :param func: callable (or a textual reference to one) to run at the given time
+        :param str|apscheduler.triggers.base.BaseTrigger trigger: trigger that determines when
+            ``func`` is called
+        :param list|tuple args: list of positional arguments to call func with
+        :param dict kwargs: dict of keyword arguments to call func with
+        :param str|unicode id: explicit identifier for the job (for modifying it later)
+        :param str|unicode name: textual description of the job
+        :param int misfire_grace_time: seconds after the designated runtime that the job is still
+            allowed to be run (or ``None`` to allow the job to run no matter how late it is)
+        :param bool coalesce: run once instead of many times if the scheduler determines that the
+            job should be run more than once in succession
+        :param int max_instances: maximum number of concurrently running instances allowed for this
+            job
+        :param datetime next_run_time: when to first run the job, regardless of the trigger (pass
+            ``None`` to add the job as paused)
+        :param str|unicode jobstore: alias of the job store to store the job in
+        :param str|unicode executor: alias of the executor to run the job with
+        :param bool replace_existing: ``True`` to replace an existing job with the same ``id``
+            (but retain the number of runs from the existing one)
+        :return: Job
+        """
         params = locals()
         params.pop("self")
         trigger_args = params.pop("trigger_args")
-        self._background_schedulers.add_job(**params, **trigger_args)
+        return self._background_schedulers.add_job(**params, **trigger_args)
 
-    def get_blocking_scheduler(self):
+    def get_blocking_scheduler(self) -> BlockingScheduler:
+        """
+        Return self blocking scheduler
+        :return: BlockingScheduler
+        """
         return self._blocking_schedulers
 
-    def get_nonblocking_scheduler(self):
+    def get_nonblocking_scheduler(self) -> BackgroundScheduler:
+        """
+        Return self background scheduler
+        :return: BackgroundScheduler
+        """
         return self._background_schedulers
 
-    def start_block_scheduler(self, *args, **kwargs):
+    def start_block_scheduler(self, *args: Any, **kwargs: Any) -> None:
+        """
+        Start blocking scheduler
+        :return: None
+        """
         self._blocking_schedulers.start(*args, **kwargs)
 
-    def start_nonblocking_scheduler(self, *args, **kwargs):
+    def start_nonblocking_scheduler(self, *args: Any, **kwargs: Any) -> None:
+        """
+        Start background scheduler
+        :return: None
+        """
         self._background_schedulers.start(*args, **kwargs)
 
-    def start_all_scheduler(self, *args, **kwargs):
+    def start_all_scheduler(self, *args: Any, **kwargs: Any) -> None:
+        """
+        Start background and blocking scheduler
+        :return: None
+        """
         self._blocking_schedulers.start(*args, **kwargs)
         self._background_schedulers.start(*args, **kwargs)
 
     def add_interval_blocking_secondly(
-            self, function: Callable, id: str = None, args: list = None,
-            kwargs: dict = None, seconds: int = 1, **trigger_args):
-        self.add_blocking_job(
+            self, function: Callable, id: str = None, args: Union[list, tuple] = None,
+            kwargs: dict = None, seconds: int = 1, **trigger_args: Any) -> Job:
+        return self.add_blocking_job(
             func=function, trigger="interval", id=id, args=args, kwargs=kwargs, seconds=seconds, **trigger_args)
 
     def add_interval_blocking_minutely(
-            self, function: Callable, id: str = None, args: list = None,
-            kwargs: dict = None, minutes: int = 1, **trigger_args):
-        self.add_blocking_job(
+            self, function: Callable, id: str = None, args: Union[list, tuple] = None,
+            kwargs: dict = None, minutes: int = 1, **trigger_args: Any) -> Job:
+        return self.add_blocking_job(
             func=function, trigger="interval", id=id, args=args, kwargs=kwargs, minutes=minutes, **trigger_args)
 
     def add_interval_blocking_hourly(
-            self, function: Callable, id: str = None, args: list = None,
-            kwargs: dict = None, hours: int = 1, **trigger_args):
-        self.add_blocking_job(
+            self, function: Callable, id: str = None, args: Union[list, tuple] = None,
+            kwargs: dict = None, hours: int = 1, **trigger_args: Any) -> Job:
+        return self.add_blocking_job(
             func=function, trigger="interval", id=id, args=args, kwargs=kwargs, hours=hours, **trigger_args)
 
     def add_interval_blocking_daily(
-            self, function: Callable, id: str = None, args: list = None,
-            kwargs: dict = None, days: int = 1, **trigger_args):
-        self.add_blocking_job(
+            self, function: Callable, id: str = None, args: Union[list, tuple] = None,
+            kwargs: dict = None, days: int = 1, **trigger_args: Any) -> Job:
+        return self.add_blocking_job(
             func=function, trigger="interval", id=id, args=args, kwargs=kwargs, days=days, **trigger_args)
 
     def add_interval_blocking_weekly(
-            self, function: Callable, id: str = None, args: list = None,
-            kwargs: dict = None, weeks: int = 1, **trigger_args):
-        self.add_blocking_job(
+            self, function: Callable, id: str = None, args: Union[list, tuple] = None,
+            kwargs: dict = None, weeks: int = 1, **trigger_args: Any) -> Job:
+        return self.add_blocking_job(
             func=function, trigger="interval", id=id, args=args, kwargs=kwargs, weeks=weeks, **trigger_args)
 
     def add_interval_nonblocking_secondly(
             self, function: Callable, id: str = None, args: list = None,
-            kwargs: dict = None, seconds: int = 1, **trigger_args):
-        self.add_nonblocking_job(
+            kwargs: dict = None, seconds: int = 1, **trigger_args: Any) -> Job:
+        return self.add_nonblocking_job(
             func=function, trigger="interval", id=id, args=args, kwargs=kwargs, seconds=seconds, **trigger_args)
 
     def add_interval_nonblocking_minutely(
             self, function: Callable, id: str = None, args: list = None,
-            kwargs: dict = None, minutes: int = 1, **trigger_args):
-        self.add_nonblocking_job(
+            kwargs: dict = None, minutes: int = 1, **trigger_args: Any) -> Job:
+        return self.add_nonblocking_job(
             func=function, trigger="interval", id=id, args=args, kwargs=kwargs, minutes=minutes, **trigger_args)
 
     def add_interval_nonblocking_hourly(
-            self, function: Callable, id: str = None, args: list = None,
-            kwargs: dict = None, hours: int = 1, **trigger_args):
-        self.add_nonblocking_job(
+            self, function: Callable, id: str = None, args: Union[list, tuple] = None,
+            kwargs: dict = None, hours: int = 1, **trigger_args: Any) -> Job:
+        return self.add_nonblocking_job(
             func=function, trigger="interval", id=id, args=args, kwargs=kwargs, hours=hours, **trigger_args)
 
     def add_interval_nonblocking_daily(
-            self, function: Callable, id: str = None, args: list = None,
-            kwargs: dict = None, days: int = 1, **trigger_args):
-        self.add_nonblocking_job(
+            self, function: Callable, id: str = None, args: Union[list, tuple] = None,
+            kwargs: dict = None, days: int = 1, **trigger_args: Any) -> Job:
+        return self.add_nonblocking_job(
             func=function, trigger="interval", id=id, args=args, kwargs=kwargs, days=days, **trigger_args)
 
     def add_interval_nonblocking_weekly(
-            self, function: Callable, id: str = None, args: list = None,
-            kwargs: dict = None, weeks: int = 1, **trigger_args):
-        self.add_nonblocking_job(
+            self, function: Callable, id: str = None, args: Union[list, tuple] = None,
+            kwargs: dict = None, weeks: int = 1, **trigger_args: Any) -> Job:
+        return self.add_nonblocking_job(
             func=function, trigger="interval", id=id, args=args, kwargs=kwargs, weeks=weeks, **trigger_args)
 
     def add_cron_blocking(
-            self, function: Callable, id: str = None, **trigger_args):
-        self.add_blocking_job(func=function, id=id, trigger="cron", **trigger_args)
+            self, function: Callable, id: str = None, **trigger_args: Any) -> Job:
+        return self.add_blocking_job(func=function, id=id, trigger="cron", **trigger_args)
 
     def add_cron_nonblocking(
-            self, function: Callable, id: str = None, **trigger_args):
-        self.add_nonblocking_job(func=function, id=id, trigger="cron", **trigger_args)
+            self, function: Callable, id: str = None, **trigger_args: Any) -> Job:
+        return self.add_nonblocking_job(func=function, id=id, trigger="cron", **trigger_args)
 
-    def remove_blocking_job(self, id: str, jobstore: str = 'default'):
-        self._blocking_schedulers.remove_job(job_id=id, jobstore=jobstore)
+    def remove_blocking_job(self, id: str, jobstore: str = 'default') -> Any:
+        return self._blocking_schedulers.remove_job(job_id=id, jobstore=jobstore)
 
-    def remove_nonblocking_job(self, id: str, jobstore: str = 'default'):
-        self._background_schedulers.remove_job(job_id=id, jobstore=jobstore)
+    def remove_nonblocking_job(self, id: str, jobstore: str = 'default') -> Any:
+        return self._background_schedulers.remove_job(job_id=id, jobstore=jobstore)
 
-    def shutdown_blocking_scheduler(self, wait: bool = False):
+    def shutdown_blocking_scheduler(self, wait: bool = False) -> None:
         self._blocking_schedulers.shutdown(wait=wait)
 
-    def shutdown_nonblocking_scheduler(self, wait: bool = False):
+    def shutdown_nonblocking_scheduler(self, wait: bool = False) -> None:
         self._background_schedulers.shutdown(wait=wait)
+
+
+scheduler_manager = SchedulerManager()

pyproject.toml

@@ -1,12 +1,12 @@
-# Rename to stable version
-# This is stable version
+# Rename to dev version
+# This is dev version
 [build-system]
 requires = ["setuptools>=61.0"]
 build-backend = "setuptools.build_meta"
 
 [project]
-name = "je_load_density"
-version = "0.0.48"
+name = "je_load_density_dev"
+version = "0.0.58"
 authors = [
     { name = "JE-Chen", email = "[email protected]" },
 ]
@@ -40,4 +40,4 @@ content-type = "text/markdown"
 license-files = ["LICENSE"]
 
 [tool.setuptools.packages]
-find = { namespaces = false }
+find = { namespaces = false }

dev.toml stable.tomldev.toml renamed to stable.toml

@@ -1,12 +1,12 @@
-# Rename to dev version
-# This is dev version
+# Rename to stable version
+# This is stable version
 [build-system]
 requires = ["setuptools>=61.0"]
 build-backend = "setuptools.build_meta"
 
 [project]
-name = "je_load_density_dev"
-version = "0.0.57"
+name = "je_load_density"
+version = "0.0.48"
 authors = [
     { name = "JE-Chen", email = "[email protected]" },
 ]