Source code for gcip2.tasks_core.task_builder

  1import dataclasses
  2import os
  3import typing
  4from collections.abc import Iterable
  5from enum import Enum
  6
  7import injector
  8import pydantic
  9
 10from gcip2 import ProjectConfig
 11from gcip2.tasks_core.models.actions import ActionBuilderImpl
 12from gcip2.tasks_core.models.params import Param, Params
 13
 14

[docs]
 15class Task(pydantic.BaseModel):
 16    basename: typing.Optional[str] = None
 17
 18    name: typing.Optional[str] = None
 19
 20    doc: str = ""
 21
 22    actions: typing.Optional[list[type["ActionBuilderImpl"]]] = None
 23
 24    parsed_params: dict[str, typing.Any] = {}
 25
 26    params: list[Param] = pydantic.Field(default_factory=lambda: [])
 27

[docs]
 28    def parse_task_params(self, args: list[str], di: injector.Injector):
 29        _config = di.get(ProjectConfig)
 30
 31        if _config.extra.model_extra:
 32            for key, val in _config.extra.model_extra.items():
 33                self.parsed_params[f"extra__{key}"] = val
 34
 35        args_iter = iter(args)
 36
 37        for arg in args_iter:
 38            if arg.startswith("--"):
 39                key = arg.removeprefix("--")
 40                param = next(
 41                    (p for p in self.params if key in [p.long, p.inverse]),
 42                    None,
 43                )
 44            elif arg.startswith("-"):
 45                key = arg.removeprefix("-")
 46                param = next(
 47                    (p for p in self.params if key in [p.short, p.inverse]),
 48                    None,
 49                )
 50            else:
 51                continue
 52
 53            if not param:
 54                raise ValueError(f"Unknown parameter: --{key}")
 55
 56            if param.type is bool:
 57                self.parsed_params[param.name] = True if key == param.long or key == param.short else False
 58                continue
 59
 60            try:
 61                value = next(args_iter)
 62            except StopIteration as err:
 63                raise StopIteration from err
 64
 65            self.parsed_params[param.name] = param.type(value)

 66

[docs]
 67    def exec_task(self, inj: injector.Injector):
 68        if not self.actions:
 69            raise RuntimeError("not found actions to execute.")
 70        for action_cls in self.actions:
 71            action = inj.create_object(action_cls)
 72            action.execute(**self.parsed_params)


 73
 74

[docs]
 75class TaskBulder:

[docs]
 76    def build(self) -> Task:
 77        raise NotImplementedError


 78
 79

[docs]
 80@injector.inject
 81@dataclasses.dataclass
 82class TaskBuilderImpl(TaskBulder):
 83    _basename: typing.ClassVar[str | Enum | None] = None
 84
 85    _params: Params
 86    _config: ProjectConfig
 87
 88    model: Task = dataclasses.field(init=False, default_factory=Task)
 89
 90    def __post_init__(self):
 91        "extension point for childs"
 92

[docs]
 93    def apply(self):
 94        return self

 95

[docs]
 96    def build(self) -> Task:
 97        if not self._basename:
 98            raise ValueError("task name not set.")
 99
100        if not self.model.doc:
101            self.model.doc = ".".join(
102                [
103                    self.__class__.__module__,
104                    self.__class__.__name__,
105                ]
106            )
107
108        if isinstance(self._basename, str):
109            self.model.basename = self._basename
110        else:
111            self.model.basename = self._basename.value
112
113        return self.model.model_copy(deep=True)

114

[docs]
115    def with_actions(self, actions: Iterable[type[ActionBuilderImpl]]):
116        self.model.actions = list(actions)
117        return self

118

[docs]
119    def with_params(self: typing.Self, params: Iterable[Param]) -> typing.Self:
120        self.model.params = list(params)
121
122        params_dict = {}
123        for param in params:
124            params_dict[param.name] = os.getenv(param.env_var, param.default) if param.env_var else param.default
125        self.model.parsed_params = params_dict
126
127        return self

128

[docs]
129    def replace_param(self, param_name: str, param: str, default: str):
130        self.model.parsed_params[param_name] = param or default

131

[docs]
132    def with_name(self, name: str):
133        self.model.name = name
134        return self

135

[docs]
136    def with_doc(self, doc: str):
137        self.model.doc = doc
138        return self