Add scheduler to executor

Add scheduler to executor

Author: JE-Chen

.idea/workspace.xml

@@ -1,7 +1,7 @@
 import builtins
 import types
 from inspect import getmembers, isbuiltin
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Union
 
 from je_auto_control.utils.exception.exception_tags import action_is_null_error, add_command_exception, \
     executor_list_error
@@ -18,6 +18,7 @@
 from je_auto_control.utils.logging.loggin_instance import auto_control_logger
 from je_auto_control.utils.package_manager.package_manager_class import package_manager
 from je_auto_control.utils.project.create_project_structure import create_project_dir
+from je_auto_control.utils.scheduler.extend_apscheduler import scheduler_manager
 from je_auto_control.utils.shell_process.shell_exec import ShellManager
 from je_auto_control.utils.start_exe.start_another_process import start_exe
 from je_auto_control.utils.test_record.record_test_class import record_action_to_list, test_record_instance
@@ -156,6 +157,16 @@ def execute_files(self, execute_files_list: list) -> List[Dict[str, str]]:
             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_auto_control/utils/executor/action_executor.py

@@ -12,6 +12,20 @@ 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: Callable, trigger: str = None, args: Union[list, tuple] = None,
@@ -196,3 +210,6 @@ def shutdown_blocking_scheduler(self, wait: bool = False) -> None:
 
     def shutdown_nonblocking_scheduler(self, wait: bool = False) -> None:
         self._background_schedulers.shutdown(wait=wait)
+
+
+scheduler_manager = SchedulerManager()

je_auto_control/utils/scheduler/extend_apscheduler.py

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "je_auto_control_dev"
-version = "0.0.87"
+version = "0.0.88"
 authors = [
     { name = "JE-Chen", email = "jechenmailman@gmail.com" },
 ]