separated the scheduler from tasks, to allow us to schedule non add url tasks in the future

app/library/EventsSubscriber.py

@@ -13,6 +13,9 @@ class Events:
     The events that can be emitted.
     """
 
+    STARTUP = "startup"
+    SHUTDOWN = "shutdown"
+
     ADDED = "added"
     UPDATED = "updated"
     COMPLETED = "completed"
@@ -42,6 +45,7 @@ class Events:
     TASK_ERROR = "task_error"
 
     PRESETS_ADD = "preset_add"
+    SCHEDULE_ADD = "schedule_add"
 
 
 @dataclass(kw_only=True)

app/library/HttpAPI.py

@@ -640,7 +640,7 @@ class HttpAPI(Common):
 
         """
         return web.json_response(
-            data=Tasks.get_instance().get_tasks(), status=web.HTTPOk.status_code, dumps=self.encoder.encode
+            data=Tasks.get_instance().get_all(), status=web.HTTPOk.status_code, dumps=self.encoder.encode
         )
 
     @route("PUT", "api/tasks")
@@ -705,7 +705,7 @@ class HttpAPI(Common):
             tasks.append(Task(**item))
 
         try:
-            tasks = ins.save(tasks=tasks).load().get_tasks()
+            tasks = ins.save(tasks=tasks).load().get_all()
         except Exception as e:
             LOG.exception(e)
             return web.json_response(

app/library/Scheduler.py

@@ -0,0 +1,136 @@
+import asyncio
+import logging
+
+from aiocron import Cron
+from aiohttp import web
+
+from .EventsSubscriber import Event, Events, EventsSubscriber
+from .Singleton import Singleton
+
+LOG = logging.getLogger("scheduler")
+
+
+class Scheduler(metaclass=Singleton):
+    """
+    This class is used to manage the schedule.
+    """
+
+    _jobs: dict[str, Cron] = {}
+    """The scheduled jobs."""
+
+    _instance = None
+    """The instance of the class."""
+
+    def __init__(self, loop: asyncio.AbstractEventLoop | None = None):
+        Scheduler._instance = self
+
+        self._loop = loop or asyncio.get_event_loop()
+
+        def handle_event(_, e: Event):
+            self.add(**e.data)
+
+        EventsSubscriber.get_instance().subscribe(Events.SCHEDULE_ADD, f"{__class__}.add", handle_event)
+
+    @staticmethod
+    def get_instance() -> "Scheduler":
+        """
+        Get the instance of the class.
+
+        Returns:
+            Scheduler: The instance of the class
+
+        """
+        if not Scheduler._instance:
+            Scheduler._instance = Scheduler()
+        return Scheduler._instance
+
+    async def on_shutdown(self, _: web.Application):
+        """
+        Do any jobs before shutdown.
+
+        Args:
+            _: The aiohttp application.
+
+        """
+        LOG.debug("Shutting down the scheduler.")
+
+        for job in self._jobs:
+            try:
+                job.stop()
+            except Exception as e:
+                LOG.exception(e)
+                LOG.error(f"Failed to stop job '{job}'. Error message '{e!s}'.")
+
+        self._jobs = {}
+
+        LOG.debug("Scheduler has been shut down.")
+
+    def attach(self, app: web.Application):
+        app.on_shutdown.append(self.on_shutdown)
+
+    def get_all(self) -> dict[str, Cron]:
+        """Return the jobs."""
+        return self._jobs
+
+    def get(self, id: str) -> Cron | None:
+        """Return the job by id."""
+        return self._jobs.get(id)
+
+    def add(self, timer: str, func: callable, args: tuple = (),
+            kwargs: dict | None = None, id: str | None = None) -> str:
+        """
+        Add a job to the schedule.
+
+        Args:
+            timer (str): The timer for the job.
+            func (callable): The function to call.
+            args (tuple): The arguments to pass to the function.
+            kwargs (dict): The keyword arguments to pass to the function.
+            id (str): The id of the job.
+
+        Returns:
+            str: The id of the job
+
+        """
+        if id and id in self._jobs:
+            self.remove(id)
+
+        job = Cron(spec=timer, func=func, args=args, kwargs=kwargs, uuid=id, start=True, loop=self._loop)
+
+        job_id = str(job.uuid)
+
+        self._jobs[job_id] = job
+
+        LOG.debug(f"Added job {job_id} to the schedule.")
+
+        return job_id
+
+    def remove(self, id: str | list[str]) -> bool:
+        """
+        Remove a job from the schedule.
+
+        Args:
+            id (str|list[str]): The id of the job to remove.
+
+        Returns:
+            bool: True if the job was removed, False otherwise
+
+        """
+        if isinstance(id, list):
+            for i in id:
+                self.remove(i)
+            return True
+
+        if id in self._jobs:
+            try:
+                self._jobs[id].stop()
+            except Exception as e:
+                LOG.exception(e)
+                LOG.error(f"Failed to stop job '{id}'. Error message '{e!s}'.")
+                return False
+
+            del self._jobs[id]
+            LOG.debug(f"Removed job {id} from the schedule.")
+            return True
+
+        return False

app/library/Tasks.py

@@ -8,13 +8,13 @@ from datetime import UTC, datetime
 from typing import Any
 
 import httpx
-from aiocron import Cron, crontab
 from aiohttp import web
 
 from .config import Config
 from .Emitter import Emitter
 from .encoder import Encoder
 from .EventsSubscriber import Event, Events, EventsSubscriber
+from .Scheduler import Scheduler
 from .Singleton import Singleton
 
 LOG = logging.getLogger("tasks")
@@ -42,21 +42,13 @@ class Task:
         return self.serialize().get(key, default)
 
 
-@dataclass(kw_only=True)
-class Job:
-    id: str
-    name: str
-    task: Task
-    job: Cron
-
-
 class Tasks(metaclass=Singleton):
     """
     This class is used to manage the tasks.
     """
 
-    _jobs: list[Job] = []
+    _tasks: list[Task] = []
-    """The jobs for the tasks."""
+    """The tasks."""
 
     _instance = None
     """The instance of the Tasks."""
@@ -69,6 +61,7 @@ class Tasks(metaclass=Singleton):
         config: Config | None = None,
         encoder: Encoder | None = None,
         client: httpx.AsyncClient | None = None,
+        scheduler: Scheduler | None = None,
     ):
         Tasks._instance = self
 
@@ -76,11 +69,12 @@ class Tasks(metaclass=Singleton):
 
         self._debug = config.debug
         self._default_preset = config.default_preset
-        self._file: str = file or os.path.join(config.config_path, "tasks.json")
+        self._file = file or os.path.join(config.config_path, "tasks.json")
-        self._client: httpx.AsyncClient = client or httpx.AsyncClient()
+        self._client = client or httpx.AsyncClient()
-        self._encoder: Encoder = encoder or Encoder()
+        self._encoder = encoder or Encoder()
-        self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_event_loop()
+        self._loop = loop or asyncio.get_event_loop()
-        self._emitter: Emitter = emitter or Emitter.get_instance()
+        self._emitter = emitter or Emitter.get_instance()
+        self._scheduler = scheduler or Scheduler.get_instance()
 
         if os.path.exists(self._file) and "600" != oct(os.stat(self._file).st_mode)[-3:]:
             try:
@@ -96,27 +90,19 @@ class Tasks(metaclass=Singleton):
     @staticmethod
     def get_instance() -> "Tasks":
         """
-        Get the instance of the Tasks.
+        Get the instance of the class.
 
         Returns:
-            Tasks: The instance of the Tasks
+            Tasks: The instance of the class.
 
         """
         if not Tasks._instance:
             Tasks._instance = Tasks()
+
         return Tasks._instance
 
     async def on_shutdown(self, _: web.Application):
-        """
+        pass
-        Shutdown the socket server.
-
-        Args:
-            _: The aiohttp application.
-
-        """
-        LOG.debug("Shutting down tasks runner.")
-        self.clear(shutdown=True)
-        LOG.debug("Tasks runner has been shut down.")
 
     def attach(self, _: web.Application):
         """
@@ -125,15 +111,12 @@ class Tasks(metaclass=Singleton):
         Args:
             _ (web.Application): The aiohttp application.
 
-        Returns:
-            None
-
         """
         self.load()
 
-    def get_tasks(self) -> list[Task]:
+    def get_all(self) -> list[Task]:
         """Return the tasks."""
-        return [job.task for job in self._jobs]
+        return self._tasks
 
     def load(self) -> "Tasks":
         """
@@ -168,18 +151,13 @@ class Tasks(metaclass=Singleton):
                 continue
 
             try:
-                self._jobs.append(
+                self._scheduler.add(timer=task.timer, func=self._runner, args=(task,), id=task.id)
-                    Job(
+                self._tasks.append(task)
-                        id=task.id,
+
-                        name=task.name,
-                        task=task,
-                        job=crontab(spec=task.timer, func=self._runner, args=(task,), start=True, loop=self._loop),
-                    )
-                )
                 LOG.info(f"Task '{i}: {task.name}' queued to be executed every '{task.timer}'.")
             except Exception as e:
                 LOG.exception(e)
-                LOG.error(f"Failed to queue task '{i}: {task['name']}'. '{e!s}'.")
+                LOG.error(f"Failed to queue task '{i}: {task.name}'. '{e!s}'.")
 
         return self
 
@@ -191,19 +169,19 @@ class Tasks(metaclass=Singleton):
             Tasks: The current instance.
 
         """
-        if len(self._jobs) < 1:
+        if len(self._tasks) < 1:
             return self
 
-        for task in self._jobs:
+        for task in self._tasks:
             try:
-                LOG.info(f"Stopping job '{task.id}: {task.name}'.")
+                LOG.info(f"Stopping task '{task.id}: {task.name}'.")
-                task.job.stop()
+                self._scheduler.remove(task.id)
             except Exception as e:
                 if not shutdown:
                     LOG.exception(e)
-                    LOG.error(f"Failed to stop job '{task.id}: {task.name}'. '{e!s}'.")
+                    LOG.error(f"Failed to stop task '{task.id}: {task.name}'. '{e!s}'.")
 
-        self._jobs.clear()
+        self._tasks.clear()
 
         return self
 

app/main.py

@@ -12,12 +12,13 @@ from aiohttp import web
 from library.config import Config
 from library.DownloadQueue import DownloadQueue
 from library.Emitter import Emitter
-from library.EventsSubscriber import EventsSubscriber
+from library.EventsSubscriber import Events, EventsSubscriber
 from library.HttpAPI import HttpAPI
 from library.HttpSocket import HttpSocket
 from library.Notifications import Notification
 from library.PackageInstaller import PackageInstaller
 from library.Presets import Presets
+from library.Scheduler import Scheduler
 from library.Tasks import Tasks
 
 LOG = logging.getLogger("app")
@@ -93,9 +94,12 @@ class Main:
         """
         Start the application.
         """
+        EventsSubscriber.get_instance().emit(Events.STARTUP, data={"app": self._app})
+
         self._socket.attach(self._app)
         self._http.attach(self._app)
         self._queue.attach(self._app)
+        Scheduler.get_instance().attach(self._app)
         Tasks.get_instance().attach(self._app)
         Presets.get_instance().attach(self._app)