Source code for gcip2.tasks_core.task_generator

 1import dataclasses
 2from collections.abc import Iterator
 3
 4import injector
 5
 6from gcip2 import ProjectConfig
 7from gcip2.tasks_core.models.params import Params
 8from gcip2.tasks_core.task_builder import Task, TaskBuilderImpl
 9
10

[docs]
11class TaskGenerator:

[docs]
12    def load_tasks(self) -> Iterator[Task]:
13        raise NotImplementedError


14
15

[docs]
16@injector.inject
17@dataclasses.dataclass
18class TaskGeneratorImpl(TaskGenerator):
19    _config: ProjectConfig
20    _di: injector.Injector
21    _params: Params
22

[docs]
23    def _auto_load_tasks(self) -> Iterator[Task]:
24        for name, cls in vars(type(self)).items():
25            if name.startswith("task_") and isinstance(cls, type) and issubclass(cls, TaskBuilderImpl):
26                yield self._di.create_object(cls).apply().build()

27

[docs]
28    def builder(self, task_class: type[TaskBuilderImpl]) -> TaskBuilderImpl:
29        return self._di.create_object(task_class)

30

[docs]
31    def load_tasks(self) -> Iterator[Task]:
32        yield from self._auto_load_tasks()