1 files changed, 449 insertions, 0 deletions
diff --git a/tools/cru-py/cru/service/_base.py b/tools/cru-py/cru/service/_base.py
new file mode 100644
index 0000000..ad813c9
--- /dev/null
+++ b/tools/cru-py/cru/service/_base.py
@@ -0,0 +1,449 @@
+from __future__ import annotations
+
+from argparse import ArgumentParser, Namespace
+from abc import ABC, abstractmethod
+import argparse
+import os
+from pathlib import Path
+from typing import TypeVar, overload
+
+from cru import CruException, CruLogicError
+
+_Feature = TypeVar("_Feature", bound="AppFeatureProvider")
+
+
+class AppError(CruException):
+    pass
+
+
+class AppFeatureError(AppError):
+    def __init__(self, message, feature: type | str, *args, **kwargs):
+        super().__init__(message, *args, **kwargs)
+        self._feature = feature
+
+    @property
+    def feature(self) -> type | str:
+        return self._feature
+
+
+class AppPathError(CruException):
+    def __init__(self, message, _path: str | Path, *args, **kwargs):
+        super().__init__(message, *args, **kwargs)
+        self._path = str(_path)
+
+    @property
+    def path(self) -> str:
+        return self._path
+
+
+class AppPath(ABC):
+    def __init__(self, id: str, is_dir: bool, description: str) -> None:
+        self._is_dir = is_dir
+        self._id = id
+        self._description = description
+
+    @property
+    @abstractmethod
+    def parent(self) -> AppPath | None: ...
+
+    @property
+    @abstractmethod
+    def app(self) -> AppBase: ...
+
+    @property
+    def id(self) -> str:
+        return self._id
+
+    @property
+    def description(self) -> str:
+        return self._description
+
+    @property
+    def is_dir(self) -> bool:
+        return self._is_dir
+
+    @property
+    @abstractmethod
+    def full_path(self) -> Path: ...
+
+    @property
+    def full_path_str(self) -> str:
+        return str(self.full_path)
+
+    def check_parents(self, must_exist: bool = False) -> bool:
+        for p in reversed(self.full_path.parents):
+            if not p.exists() and not must_exist:
+                return False
+            if not p.is_dir():
+                raise AppPathError("Parents' path must be a dir.", self.full_path)
+        return True
+
+    def check_self(self, must_exist: bool = False) -> bool:
+        if not self.check_parents(must_exist):
+            return False
+        if not self.full_path.exists():
+            if not must_exist:
+                return False
+            raise AppPathError("Not exist.", self.full_path)
+        if self.is_dir:
+            if not self.full_path.is_dir():
+                raise AppPathError("Should be a directory, but not.", self.full_path)
+            else:
+                return True
+        else:
+            if not self.full_path.is_file():
+                raise AppPathError("Should be a file, but not.", self.full_path)
+            else:
+                return True
+
+    def ensure(self, create_file: bool = False) -> None:
+        e = self.check_self(False)
+        if not e:
+            os.makedirs(self.full_path.parent, exist_ok=True)
+            if self.is_dir:
+                os.mkdir(self.full_path)
+            elif create_file:
+                with open(self.full_path, "w") as f:
+                    f.write("")
+
+    def add_subpath(
+        self,
+        name: str,
+        is_dir: bool,
+        /,
+        id: str | None = None,
+        description: str = "",
+    ) -> AppFeaturePath:
+        return self.app.add_path(name, is_dir, self, id, description)
+
+    @property
+    def app_relative_path(self) -> Path:
+        return self.full_path.relative_to(self.app.root.full_path)
+
+
+class AppFeaturePath(AppPath):
+    def __init__(
+        self,
+        parent: AppPath,
+        name: str,
+        is_dir: bool,
+        /,
+        id: str | None = None,
+        description: str = "",
+    ) -> None:
+        super().__init__(id or name, is_dir, description)
+        self._name = name
+        self._parent = parent
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def parent(self) -> AppPath:
+        return self._parent
+
+    @property
+    def app(self) -> AppBase:
+        return self.parent.app
+
+    @property
+    def full_path(self) -> Path:
+        return Path(self.parent.full_path, self.name).resolve()
+
+
+class AppRootPath(AppPath):
+    def __init__(self, app: AppBase):
+        super().__init__("root", True, "Application root path.")
+        self._app = app
+        self._full_path: Path | None = None
+
+    @property
+    def parent(self) -> None:
+        return None
+
+    @property
+    def app(self) -> AppBase:
+        return self._app
+
+    @property
+    def full_path(self) -> Path:
+        if self._full_path is None:
+            raise AppError("App root path is not set yet.")
+        return self._full_path
+
+    def setup(self, path: os.PathLike) -> None:
+        if self._full_path is not None:
+            raise AppError("App root path is already set.")
+        self._full_path = Path(path).resolve()
+
+
+class AppFeatureProvider(ABC):
+    def __init__(self, name: str, /, app: AppBase | None = None):
+        super().__init__()
+        self._name = name
+        self._app = app if app else AppBase.get_instance()
+
+    @property
+    def app(self) -> AppBase:
+        return self._app
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @abstractmethod
+    def setup(self) -> None: ...
+
+
+class AppCommandFeatureProvider(AppFeatureProvider):
+    @abstractmethod
+    def get_command_info(self) -> tuple[str, str]: ...
+
+    @abstractmethod
+    def setup_arg_parser(self, arg_parser: ArgumentParser): ...
+
+    @abstractmethod
+    def run_command(self, args: Namespace) -> None: ...
+
+
+DATA_DIR_NAME = "data"
+
+
+class PathCommandProvider(AppCommandFeatureProvider):
+    def __init__(self) -> None:
+        super().__init__("path-command-provider")
+
+    def setup(self):
+        pass
+
+    def get_command_info(self):
+        return ("path", "Get information about paths used by app.")
+
+    def setup_arg_parser(self, arg_parser: ArgumentParser) -> None:
+        subparsers = arg_parser.add_subparsers(
+            dest="path_command", required=True, metavar="PATH_COMMAND"
+        )
+        _list_parser = subparsers.add_parser(
+            "list", help="list special paths used by app"
+        )
+
+    def run_command(self, args: Namespace) -> None:
+        if args.path_command == "list":
+            for path in self.app.paths:
+                print(f"{path.app_relative_path.as_posix()}: {path.description}")
+
+
+class CommandDispatcher(AppFeatureProvider):
+    def __init__(self) -> None:
+        super().__init__("command-dispatcher")
+        self._parsed_args: argparse.Namespace | None = None
+
+    def setup_arg_parser(self) -> None:
+        epilog = """
+==> to start,
+./tools/manage init
+./tools/manage config init
+ln -s generated/docker-compose.yaml .
+# Then edit config file.
+
+==> to update
+git pull
+./tools/manage template generate --no-dry-run
+docker compose up
+        """.strip()
+
+        self._map: dict[str, AppCommandFeatureProvider] = {}
+        arg_parser = argparse.ArgumentParser(
+            description="Service management",
+            formatter_class=argparse.RawDescriptionHelpFormatter,
+            epilog=epilog,
+        )
+        arg_parser.add_argument(
+            "--project-dir",
+            help="The path of the project directory.",
+            required=True,
+            type=str,
+        )
+        subparsers = arg_parser.add_subparsers(
+            dest="command",
+            help="The management command to execute.",
+            metavar="COMMAND",
+        )
+        for feature in self.app.features:
+            if isinstance(feature, AppCommandFeatureProvider):
+                info = feature.get_command_info()
+                command_subparser = subparsers.add_parser(info[0], help=info[1])
+                feature.setup_arg_parser(command_subparser)
+                self._map[info[0]] = feature
+        self._arg_parser = arg_parser
+
+    def setup(self):
+        pass
+
+    @property
+    def arg_parser(self) -> argparse.ArgumentParser:
+        return self._arg_parser
+
+    @property
+    def map(self) -> dict[str, AppCommandFeatureProvider]:
+        return self._map
+
+    def get_program_parsed_args(self) -> argparse.Namespace:
+        if self._parsed_args is None:
+            self._parsed_args = self.arg_parser.parse_args()
+        return self._parsed_args
+
+    def run_command(self, args: argparse.Namespace | None = None) -> None:
+        real_args = args or self.get_program_parsed_args()
+        if real_args.command is None:
+            self.arg_parser.print_help()
+            return
+        self.map[real_args.command].run_command(real_args)
+
+
+class AppInitializer(AppCommandFeatureProvider):
+    def __init__(self) -> None:
+        super().__init__("app-initializer")
+
+    def _init_app(self) -> bool:
+        if self.app.app_initialized:
+            return False
+        self.app.data_dir.ensure()
+        return True
+
+    def setup(self):
+        pass
+
+    def get_command_info(self):
+        return ("init", "Initialize the app.")
+
+    def setup_arg_parser(self, arg_parser):
+        pass
+
+    def run_command(self, args):
+        init = self._init_app()
+        if init:
+            print("App initialized successfully.")
+        else:
+            print("App is already initialized. Do nothing.")
+
+
+class AppBase:
+    _instance: AppBase | None = None
+
+    @staticmethod
+    def get_instance() -> AppBase:
+        if AppBase._instance is None:
+            raise AppError("App instance not initialized")
+        return AppBase._instance
+
+    def __init__(self, app_id: str, name: str):
+        AppBase._instance = self
+        self._app_id = app_id
+        self._name = name
+        self._root = AppRootPath(self)
+        self._paths: list[AppFeaturePath] = []
+        self._features: list[AppFeatureProvider] = []
+
+    def setup(self) -> None:
+        command_dispatcher = self.get_feature(CommandDispatcher)
+        command_dispatcher.setup_arg_parser()
+        program_args = command_dispatcher.get_program_parsed_args()
+        self.setup_root(program_args.project_dir)
+        self._data_dir = self.add_path(DATA_DIR_NAME, True, id="data")
+        for feature in self.features:
+            feature.setup()
+        for path in self.paths:
+            path.check_self()
+
+    @property
+    def app_id(self) -> str:
+        return self._app_id
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def root(self) -> AppRootPath:
+        return self._root
+
+    def setup_root(self, path: os.PathLike) -> None:
+        self._root.setup(path)
+
+    @property
+    def data_dir(self) -> AppFeaturePath:
+        return self._data_dir
+
+    @property
+    def app_initialized(self) -> bool:
+        return self.data_dir.check_self()
+
+    def ensure_app_initialized(self) -> AppRootPath:
+        if not self.app_initialized:
+            raise AppError(
+                user_message="Root directory does not exist. "
+                "Please run 'init' to create one."
+            )
+        return self.root
+
+    @property
+    def features(self) -> list[AppFeatureProvider]:
+        return self._features
+
+    @property
+    def paths(self) -> list[AppFeaturePath]:
+        return self._paths
+
+    def add_feature(self, feature: _Feature) -> _Feature:
+        for f in self.features:
+            if f.name == feature.name:
+                raise AppFeatureError(
+                    f"Duplicate feature name: {feature.name}.", feature.name
+                )
+        self._features.append(feature)
+        return feature
+
+    def add_path(
+        self,
+        name: str,
+        is_dir: bool,
+        /,
+        parent: AppPath | None = None,
+        id: str | None = None,
+        description: str = "",
+    ) -> AppFeaturePath:
+        p = AppFeaturePath(
+            parent or self.root, name, is_dir, id=id, description=description
+        )
+        self._paths.append(p)
+        return p
+
+    @overload
+    def get_feature(self, feature: str) -> AppFeatureProvider: ...
+
+    @overload
+    def get_feature(self, feature: type[_Feature]) -> _Feature: ...
+
+    def get_feature(
+        self, feature: str | type[_Feature]
+    ) -> AppFeatureProvider | _Feature:
+        if isinstance(feature, str):
+            for f in self._features:
+                if f.name == feature:
+                    return f
+        elif isinstance(feature, type):
+            for f in self._features:
+                if isinstance(f, feature):
+                    return f
+        else:
+            raise CruLogicError("Argument must be the name of feature or its class.")
+
+        raise AppFeatureError(f"Feature {feature} not found.", feature)
+
+    def get_path(self, name: str) -> AppFeaturePath:
+        for p in self._paths:
+            if p.id == name or p.name == name:
+                return p
+        raise AppPathError(f"Application path {name} not found.", name)