Mercurial > libervia-web
diff libervia/server/tasks/manager.py @ 1261:a46d0e0f383b
tasks: `AFTER` attribute to handle tasks order:
A Task instance can now put a list of tasks names in its `AFTER` attribute, if will then
be prepared and launched after it.
author | Goffi <goffi@goffi.org> |
---|---|
date | Sun, 03 May 2020 21:07:23 +0200 |
parents | 6161076193e0 |
children | df40708c4c76 |
line wrap: on
line diff
--- a/libervia/server/tasks/manager.py Sun May 03 21:07:22 2020 +0200 +++ b/libervia/server/tasks/manager.py Sun May 03 21:07:23 2020 +0200 @@ -18,6 +18,7 @@ import os import os.path from pathlib import Path +from typing import Dict import importlib.util from twisted.internet import defer from sat.core.log import getLogger @@ -26,6 +27,7 @@ from sat.tools import utils from libervia.server.constants import Const as C from . import implicit +from .task import Task log = getLogger(__name__) @@ -81,9 +83,61 @@ raise ValueError(_("Unexpected value for {var}: {value!r}").format( var=var, value=value)) - async def parseTasksDir(self, dir_path): + async def importTask( + self, + task_name: str, + task_path: Path, + to_import: Dict[str, Path] + ) -> None: + if task_name in self.tasks: + log.debug(f"skipping task {task_name} which is already imported") + return + module_name = f"{self.site_name or C.SITE_NAME_DEFAULT}.task.{task_name}" + + spec = importlib.util.spec_from_file_location(module_name, task_path) + task_module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(task_module) + task = task_module.Task(self, task_name) + if task.AFTER is not None: + for pre_task_name in task.AFTER: + log.debug( + f"task {task_name!r} must be run after {pre_task_name!r}") + try: + pre_task_path = to_import[pre_task_name] + except KeyError: + raise ValueError( + f"task {task_name!r} must be run after {pre_task_name!r}, " + f"however there is no task with such name") + await self.importTask(pre_task_name, pre_task_path, to_import) + + # we launch prepare, which is a method used to prepare + # data at runtime (e.g. set WATCH_DIRS using config) + try: + prepare = task.prepare + except AttributeError: + pass + else: + log.info(_('== preparing task "{task_name}" for {site_name} =='.format( + task_name=task_name, site_name=self.site_name or DEFAULT_SITE_LABEL))) + try: + await utils.asDeferred(prepare) + except exceptions.CancelError as e: + log.debug(f"Skipping {task_name} which cancelled itself: {e}") + return + + self.tasks[task_name] = task + self.validateData(task) + if self.host.options['dev_mode']: + dirs = task.WATCH_DIRS or [] + for dir_ in dirs: + self.host.files_watcher.watchDir( + dir_, auto_add=True, recursive=True, + callback=self._autorunTask, task_name=task_name) + + async def parseTasksDir(self, dir_path: Path) -> None: log.debug(f"parsing tasks in {dir_path}") tasks_paths = sorted(dir_path.glob('task_*.py')) + to_import = {} for task_path in tasks_paths: if not task_path.is_file(): continue @@ -95,34 +149,10 @@ "A task with the name [{name}] already exists".format( name=task_name)) log.debug(f"task {task_name} found") - module_name = f"{self.site_name or C.SITE_NAME_DEFAULT}.task.{task_name}" - - spec = importlib.util.spec_from_file_location(module_name, task_path) - task_module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(task_module) - task = task_module.Task(self) + to_import[task_name] = task_path - # we launch prepare, which is a method used to prepare - # data at runtime (e.g. set WATCH_DIRS using config) - try: - prepare = task.prepare - except AttributeError: - pass - else: - try: - await utils.asDeferred(prepare) - except exceptions.CancelError as e: - log.debug(f"Skipping {task_name} which cancelled itself: {e}") - continue - - self.tasks[task_name] = task - self.validateData(task) - if self.host.options['dev_mode']: - dirs = task.WATCH_DIRS or [] - for dir_ in dirs: - self.host.files_watcher.watchDir( - dir_, auto_add=True, recursive=True, - callback=self._autorunTask, task_name=task_name) + for task_name, task_path in to_import.items(): + await self.importTask(task_name, task_path, to_import) async def parseTasks(self): # implicit tasks are always run