Source code for gcip2.tasks_core.models.actions

 1import dataclasses
 2import shlex
 3import subprocess
 4import typing
 5from abc import ABC, abstractmethod
 6from collections.abc import Iterable
 7
 8import injector
 9
10from gcip2.logging import logger as LOGGER
11from gcip2.project_config import ProjectConfig
12from gcip2.vault import SecretsHandler
13
14ActionResult = typing.Any
15
16

[docs]
17@injector.inject
18@dataclasses.dataclass
19class ActionBuilderImpl(ABC):
20    _di: injector.Injector
21    _config: ProjectConfig
22    _secret_handler: SecretsHandler
23

[docs]
24    @abstractmethod
25    def impl(self, **kwargs: typing.Any) -> ActionResult: ...

26

[docs]
27    @abstractmethod
28    def execute(self, **kwargs: typing.Any): ...


29
30

[docs]
31class InteractivePythonAction(ActionBuilderImpl):

[docs]
32    @typing.override
33    def impl(self, **kwargs: typing.Any) -> None:
34        raise NotImplementedError

35

[docs]
36    @typing.override
37    def execute(self, **kwargs: typing.Any) -> None:
38        LOGGER.debug(f"running InteractivePythonAction action: {self.__class__.__name__}")
39        self.impl(**kwargs)


40
41

[docs]
42class InteractiveShlex(ActionBuilderImpl):

[docs]
43    @typing.override
44    def impl(self, **kwargs: typing.Any) -> Iterable[list[str]]:
45        raise NotImplementedError

46

[docs]
47    @typing.override
48    def execute(self, **kwargs: typing.Any) -> None:
49        LOGGER.debug(f"running InteractiveShlex action: {self.__class__.__name__}")
50
51        commands = [shlex.join(cmd) for cmd in self.impl(**kwargs)]
52
53        script = " && ".join(commands)
54
55        LOGGER.info("running: {}", script)
56        subprocess.run(script, shell=True, check=True)