Source code for gcip2.tasks_core.registry

 1import importlib.util
 2import pathlib
 3import sys
 4
 5import injector
 6
 7from gcip2.logging import logger as LOGGER
 8from gcip2.project_config import ProjectConfig
 9from gcip2.tasks_core.ci_tasks._tasks import CiTaskGenerator
10from gcip2.tasks_core.task_builder import Task
11from gcip2.tasks_core.task_generator import TaskGenerator
12
13

[docs]
14class TaskRegistry:
15    @injector.inject
16    def __init__(self, config: ProjectConfig, di: injector.Injector) -> None:
17        self._di = di
18        self._config = config
19
20        self._tasks: dict[str, Task] | None = None
21

[docs]
22    def _load_module(self, module: str):
23
24        module_path = pathlib.Path(module)
25
26        LOGGER.debug(f"Tasks loaded from module: {module_path}")
27
28        if module_path.is_file():
29            module_name = module_path.stem
30
31            spec = importlib.util.spec_from_file_location(module_name, module_path)
32            if spec is None or spec.loader is None:
33                raise ImportError(f"Could not load module from {module_path}")
34
35            module_obj = importlib.util.module_from_spec(spec)
36            sys.modules[module_name] = module_obj
37            spec.loader.exec_module(module_obj)
38            return module_obj
39
40        sys.path.insert(0, str(module_path.parent))
41
42        try:
43            module_obj = __import__(module, fromlist=["TaskGenerator"])
44        finally:
45            sys.path.remove(str(module_path.parent))
46
47        return module_obj

48

[docs]
49    def _load_tasks_from_generator(self, generator: TaskGenerator) -> dict[str, Task]:
50        registry: dict[str, Task] = {}
51
52        for task in generator.load_tasks():
53            if not task.basename:
54                raise ValueError("Task basename is not set.")
55            taskname = task.basename if not task.name else ":".join([task.basename, task.name])
56            registry[taskname] = task
57
58        return registry

59

[docs]
60    def load_tasks_from_module(self, module: str | None) -> dict[str, Task]:
61        if not module:
62            LOGGER.warning("module with tasks not found")
63            return {}
64
65        module_obj = self._load_module(module)
66
67        generator: TaskGenerator = self._di.create_object(module_obj.TaskGenerator)
68
69        registry: dict[str, Task] = self._load_tasks_from_generator(generator)
70
71        return registry

72
73    @property
74    def tasks(self) -> dict[str, Task]:
75        if self._tasks is None:
76            self._tasks = {
77                **self._load_tasks_from_generator(self._di.create_object(CiTaskGenerator)),
78                **self.load_tasks_from_module(self._config.extra.system.tasks_module),
79            }
80        return self._tasks