aboutsummaryrefslogtreecommitdiff
path: root/tools/cru-py/cru/service/_base.py
blob: c84ab05f3706e655e1c6460c9d194c81e7b18e9e (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
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 InternalAppException(CruInternalError):
    pass


class AppFeatureProvider:
    def __init__(self, name: str, /, app: AppBase | None = None):
        super().__init__()
        self._name = name
        self._app = app if app else AppBase.get_instance()
        self._app_paths: list[ApplicationPath] = []
        self.app.add_app_feature(self)

    @property
    def app(self) -> AppBase:
        return self._app

    @property
    def name(self) -> str:
        return self._name

    @property
    def app_paths(self) -> list[ApplicationPath]:
        return self._app_paths

    def add_app_path(self, subpath: str, is_dir: bool) -> ApplicationPath:
        p = ApplicationPath(self.app.app_dir, subpath, is_dir)
        self._app_paths.append(p)
        return p


class AppCommandFeatureProvider(AppFeatureProvider, ABC):
    @abstractmethod
    def add_arg_parser(self, arg_parser: ArgumentParser) -> None: ...

    @abstractmethod
    def run_command(self, args: Namespace) -> None: ...


class AppBase(CruApplication):
    _instance: AppBase | None = None

    @staticmethod
    def get_instance() -> AppBase:
        if AppBase._instance is None:
            raise CruInternalError("App instance not initialized")
        return AppBase._instance

    def __init__(self, name: str, app_dir: str):
        super().__init__(name)
        AppBase._instance = self
        self._app_dir = app_dir
        self._app_features: list[AppFeatureProvider] = []

    @property
    def app_dir(self) -> str:
        return self._app_dir

    @property
    def app_features(self) -> list[AppFeatureProvider]:
        return self._app_features

    @property
    def app_paths(self) -> Iterable[ApplicationPath]:
        return (
            CruIterator(self._app_features).transform(lambda x: x.app_paths).flatten(1)
        )

    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.")