diff options
| -rw-r--r-- | tools/cru-py/cru/service/_base.py | 33 | ||||
| -rw-r--r-- | tools/cru-py/cru/service/_config.py | 15 | ||||
| -rw-r--r-- | tools/cru-py/cru/service/_data.py | 12 | ||||
| -rw-r--r-- | tools/cru-py/cru/service/_docker.py | 19 | ||||
| -rw-r--r-- | tools/cru-py/cru/service/_template.py | 4 | ||||
| -rw-r--r-- | tools/cru-py/cru/tool.py | 82 |
6 files changed, 150 insertions, 15 deletions
diff --git a/tools/cru-py/cru/service/_base.py b/tools/cru-py/cru/service/_base.py index 7bb09ad..c84ab05 100644 --- a/tools/cru-py/cru/service/_base.py +++ b/tools/cru-py/cru/service/_base.py @@ -3,12 +3,19 @@ from __future__ import annotations from argparse import ArgumentParser, Namespace from abc import ABC, abstractmethod from collections.abc import Iterable +from typing import TypeVar, overload from cru import CruIterator, CruInternalError from cru.app import ApplicationPath, CruApplication +_F = TypeVar("_F") -class AppFeatureProvider(ABC): + +class InternalAppException(CruInternalError): + pass + + +class AppFeatureProvider: def __init__(self, name: str, /, app: AppBase | None = None): super().__init__() self._name = name @@ -33,6 +40,8 @@ class AppFeatureProvider(ABC): self._app_paths.append(p) return p + +class AppCommandFeatureProvider(AppFeatureProvider, ABC): @abstractmethod def add_arg_parser(self, arg_parser: ArgumentParser) -> None: ... @@ -71,3 +80,25 @@ class AppBase(CruApplication): def add_app_feature(self, feature: AppFeatureProvider) -> None: self._app_features.append(feature) + + @overload + def get_feature(self, feature: str) -> AppFeatureProvider: ... + + @overload + def get_feature(self, feature: type[_F]) -> _F: ... + + def get_feature(self, feature: str | type[_F]) -> AppFeatureProvider | _F: + if isinstance(feature, str): + for f in self._app_features: + if f.name == feature: + return f + elif isinstance(feature, type): + for f in self._app_features: + if isinstance(f, feature): + return f + else: + raise InternalAppException( + "Argument must be the name of feature or its class." + ) + + raise InternalAppException(f"Feature {feature} not found.") diff --git a/tools/cru-py/cru/service/_config.py b/tools/cru-py/cru/service/_config.py new file mode 100644 index 0000000..1838015 --- /dev/null +++ b/tools/cru-py/cru/service/_config.py @@ -0,0 +1,15 @@ +import os.path +from ._base import AppFeatureProvider +from ._data import DataManager + + +class ConfigManager(AppFeatureProvider): + def __init__(self, config_file_name="config") -> None: + super().__init__("config-manager") + self._file_name = config_file_name + + @property + def config_file_path(self) -> str: + return os.path.join( + self.app.get_feature(DataManager).data_dir.full_path, self._file_name + ) diff --git a/tools/cru-py/cru/service/_data.py b/tools/cru-py/cru/service/_data.py new file mode 100644 index 0000000..f38dc23 --- /dev/null +++ b/tools/cru-py/cru/service/_data.py @@ -0,0 +1,12 @@ +from cru.app import ApplicationPath +from ._base import AppFeatureProvider + + +class DataManager(AppFeatureProvider): + def __init__(self) -> None: + super().__init__("data-manager") + self._dir = self.add_app_path("data", True) + + @property + def data_dir(self) -> ApplicationPath: + return self._dir diff --git a/tools/cru-py/cru/service/_docker.py b/tools/cru-py/cru/service/_docker.py index 5958f4f..9b801c4 100644 --- a/tools/cru-py/cru/service/_docker.py +++ b/tools/cru-py/cru/service/_docker.py @@ -1,24 +1,19 @@ -import shutil import subprocess -from .._util import L +from cru.tool import ExternalTool -class DockerController: +class DockerController(ExternalTool): DOCKER_BIN_NAME = "docker" def __init__(self, docker_bin: None | str = None) -> None: - self._docker_bin = docker_bin - - @property - def docker_bin(self) -> str: - if self._docker_bin is None: - self._docker_bin = shutil.which(self.DOCKER_BIN_NAME) - return self._docker_bin + super().__init__(docker_bin or self.DOCKER_BIN_NAME) def list_containers(self) -> L[str]: - p = subprocess.run([self.docker_bin, "container", "ls", ""], capture_output=True) + p = subprocess.run( + [self.docker_bin, "container", "ls", ""], capture_output=True + ) return p.stdout.decode("utf-8").splitlines() def restart_container(self, container_name: str) -> None: - subprocess.run([self.docker_bin, "restart", container_name])
\ No newline at end of file + subprocess.run([self.docker_bin, "restart", container_name]) diff --git a/tools/cru-py/cru/service/_template.py b/tools/cru-py/cru/service/_template.py index 5da00ba..bf13212 100644 --- a/tools/cru-py/cru/service/_template.py +++ b/tools/cru-py/cru/service/_template.py @@ -1,12 +1,12 @@ from argparse import ArgumentParser, Namespace -from ._base import AppFeatureProvider +from ._base import AppCommandFeatureProvider from cru.app import ApplicationPath from cru.template import TemplateTree -class TemplateManager(AppFeatureProvider): +class TemplateManager(AppCommandFeatureProvider): def __init__(self, prefix: str = "CRUPEST"): super().__init__("template-manager") self._templates_dir = self.add_app_path("templates", True) diff --git a/tools/cru-py/cru/tool.py b/tools/cru-py/cru/tool.py new file mode 100644 index 0000000..377f5d7 --- /dev/null +++ b/tools/cru-py/cru/tool.py @@ -0,0 +1,82 @@ +import shutil +import subprocess +from typing import Any +from collections.abc import Iterable + +from ._error import CruException + + +class CruExternalToolError(CruException): + def __init__(self, message: str, tool: str, *args, **kwargs) -> None: + super().__init__(message, *args, **kwargs) + self._tool = tool + + @property + def tool(self) -> str: + return self._tool + + +class CruExternalToolNotFoundError(CruExternalToolError): + def __init__(self, message: str | None, tool: str, *args, **kwargs) -> None: + super().__init__( + message or f"Could not find binary for {tool}.", tool, *args, **kwargs + ) + + +class CruExternalToolRunError(CruExternalToolError): + def __init__( + self, + message: str, + tool: str, + tool_args: Iterable[str], + tool_error: Any, + *args, + **kwargs, + ) -> None: + super().__init__(message, tool, *args, **kwargs) + self._tool_args = list(tool_args) + self._tool_error = tool_error + + @property + def tool_args(self) -> list[str]: + return self._tool_args + + @property + def tool_error(self) -> Any: + return self._tool_error + + +class ExternalTool: + def __init__(self, bin: str) -> None: + self._bin = bin + + @property + def bin(self) -> str: + return self._bin + + @bin.setter + def bin(self, value: str) -> None: + self._bin = value + + @property + def bin_path(self) -> str: + real_bin = shutil.which(self.bin) + if not real_bin: + raise CruExternalToolNotFoundError(None, self.bin) + return real_bin + + def run( + self, *process_args: str, **subprocess_kwargs + ) -> subprocess.CompletedProcess: + try: + return subprocess.run( + [self.bin_path] + list(process_args), **subprocess_kwargs + ) + except subprocess.CalledProcessError as e: + raise CruExternalToolError("Subprocess failed.", self.bin) from e + except OSError as e: + raise CruExternalToolError("Failed to start subprocess", self.bin) from e + + def run_get_output(self, *process_args: str, **subprocess_kwargs) -> Any: + process = self.run(*process_args, capture_output=True, **subprocess_kwargs) + return process.stdout |