From d88ee150604801146c1640c3d9e5976040767616 Mon Sep 17 00:00:00 2001 From: ArabCoders Date: Sun, 2 Mar 2025 21:49:52 +0300 Subject: [PATCH] separated the scheduler from tasks, to allow us to schedule non add url tasks in the future --- app/library/EventsSubscriber.py | 4 + app/library/HttpAPI.py | 4 +- app/library/Scheduler.py | 136 ++++++++++++++++++++++++++++++++ app/library/Tasks.py | 74 ++++++----------- app/main.py | 6 +- 5 files changed, 173 insertions(+), 51 deletions(-) create mode 100644 app/library/Scheduler.py diff --git a/app/library/EventsSubscriber.py b/app/library/EventsSubscriber.py index 45d8b659..21e96adc 100644 --- a/app/library/EventsSubscriber.py +++ b/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) diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py index afaa3390..43a982c8 100644 --- a/app/library/HttpAPI.py +++ b/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( diff --git a/app/library/Scheduler.py b/app/library/Scheduler.py new file mode 100644 index 00000000..215bb1d8 --- /dev/null +++ b/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 diff --git a/app/library/Tasks.py b/app/library/Tasks.py index e8c56555..0550590b 100644 --- a/app/library/Tasks.py +++ b/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] = [] - """The jobs for the tasks.""" + _tasks: list[Task] = [] + """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._client: httpx.AsyncClient = client or httpx.AsyncClient() - self._encoder: Encoder = encoder or Encoder() - self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_event_loop() - self._emitter: Emitter = emitter or Emitter.get_instance() + self._file = file or os.path.join(config.config_path, "tasks.json") + self._client = client or httpx.AsyncClient() + self._encoder = encoder or Encoder() + self._loop = loop or asyncio.get_event_loop() + 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): - """ - 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.") + pass 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( - Job( - id=task.id, - name=task.name, - task=task, - job=crontab(spec=task.timer, func=self._runner, args=(task,), start=True, loop=self._loop), - ) - ) + self._scheduler.add(timer=task.timer, func=self._runner, args=(task,), id=task.id) + self._tasks.append(task) + 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}'.") - task.job.stop() + LOG.info(f"Stopping task '{task.id}: {task.name}'.") + 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 diff --git a/app/main.py b/app/main.py index 10c4b658..8262428f 100644 --- a/app/main.py +++ b/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)