diff options
| -rw-r--r-- | tools/cru-py/cru/config.py | 14 | ||||
| -rw-r--r-- | tools/cru-py/cru/list.py | 10 | ||||
| -rw-r--r-- | tools/cru-py/cru/service/_app.py | 5 | ||||
| -rw-r--r-- | tools/cru-py/cru/service/_base.py | 4 | ||||
| -rw-r--r-- | tools/cru-py/cru/service/_config.py | 116 | ||||
| -rw-r--r-- | tools/cru-py/cru/service/_nginx.py | 74 | ||||
| -rw-r--r-- | tools/cru-py/cru/service/_template.py | 16 | ||||
| -rw-r--r-- | tools/cru-py/cru/template.py | 4 |
8 files changed, 154 insertions, 89 deletions
diff --git a/tools/cru-py/cru/config.py b/tools/cru-py/cru/config.py index 9efcd55..0f6f0d0 100644 --- a/tools/cru-py/cru/config.py +++ b/tools/cru-py/cru/config.py @@ -113,7 +113,7 @@ class ConfigItem(Generic[_T]): self.value, self.default, ) - + @property def description_str(self) -> str: return f"{self.name} ({self.value_type.name}): {self.description}" @@ -123,9 +123,19 @@ class Configuration(CruUniqueKeyList[ConfigItem[Any], str]): def __init__(self): super().__init__(lambda c: c.name) + def get_set_items(self) -> list[ConfigItem[Any]]: + return [item for item in self if item.is_set] + + def get_unset_items(self) -> list[ConfigItem[Any]]: + return [item for item in self if not item.is_set] + + @property + def all_set(self) -> bool: + return len(self.get_unset_items()) == 0 + @property def all_not_set(self) -> bool: - return self.cru_iter().all(lambda item: not item.is_set) + return len(self.get_set_items()) == 0 def add_text_config( self, diff --git a/tools/cru-py/cru/list.py b/tools/cru-py/cru/list.py index e329ae2..9d210b7 100644 --- a/tools/cru-py/cru/list.py +++ b/tools/cru-py/cru/list.py @@ -1,7 +1,7 @@ from __future__ import annotations from collections.abc import Callable, Iterator -from typing import Any, Generic, Iterable, TypeAlias, TypeVar +from typing import Any, Generic, Iterable, TypeAlias, TypeVar, overload from ._error import CruInternalError from ._iter import CruIterator @@ -78,6 +78,14 @@ class CruUniqueKeyList(Generic[_T, _K]): if len(keys) != len(set(keys)): raise CruInternalError("Duplicate keys!") + @overload + def get_or( + self, key: _K, fallback: CruNotFound = CruNotFound.VALUE + ) -> _T | CruNotFound: ... + + @overload + def get_or(self, key: _K, fallback: _O) -> _T | _O: ... + def get_or( self, key: _K, fallback: _O | CruNotFound = CruNotFound.VALUE ) -> _T | _O | CruNotFound: diff --git a/tools/cru-py/cru/service/_app.py b/tools/cru-py/cru/service/_app.py index d6f74dd..e72baec 100644 --- a/tools/cru-py/cru/service/_app.py +++ b/tools/cru-py/cru/service/_app.py @@ -7,17 +7,20 @@ from ._base import ( from ._config import ConfigManager from ._data import DataManager from ._template import TemplateManager +from ._nginx import NginxManager APP_ID = "crupest" + class App(AppBase): def __init__(self): - super().__init__(APP_ID,f"{APP_ID}-service") + super().__init__(APP_ID, f"{APP_ID}-service") self.add_feature(PathCommandProvider()) self.add_feature(AppInitializer()) self.add_feature(DataManager()) self.add_feature(ConfigManager()) self.add_feature(TemplateManager()) + self.add_feature(NginxManager()) self.add_feature(CommandDispatcher()) def run_command(self): diff --git a/tools/cru-py/cru/service/_base.py b/tools/cru-py/cru/service/_base.py index aecd7c9..90e1f85 100644 --- a/tools/cru-py/cru/service/_base.py +++ b/tools/cru-py/cru/service/_base.py @@ -219,7 +219,9 @@ class PathCommandProvider(AppCommandFeatureProvider): subparsers = arg_parser.add_subparsers( dest="path_command", required=True, metavar="PATH_COMMAND" ) - _list_parser = subparsers.add_parser("list", help="list all paths.") + _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": diff --git a/tools/cru-py/cru/service/_config.py b/tools/cru-py/cru/service/_config.py index c28b071..52fed34 100644 --- a/tools/cru-py/cru/service/_config.py +++ b/tools/cru-py/cru/service/_config.py @@ -1,5 +1,5 @@ from collections.abc import Iterable -from typing import Any +from typing import Any, Literal, overload from cru import CruException from cru.config import Configuration, ConfigItem @@ -117,41 +117,6 @@ class AppConfigDuplicateEntryError(AppConfigFileEntryError): return "Duplicate entries found in config file" -class AppConfigEntryKeyError(AppConfigFileEntryError): - def __init__( - self, - message: str, - configuration: Configuration, - undefined_entries: Iterable[SimpleLineConfigParser.Entry], - unset_items: Iterable[ConfigItem], - *args, - **kwargs, - ) -> None: - super().__init__(message, configuration, undefined_entries, *args, **kwargs) - self._unset_items = list(unset_items) - - @property - def unset_items(self) -> list[ConfigItem]: - return self._unset_items - - @property - def friendly_message_head(self) -> str: - return "Entry key not defined in app config" - - @property - def unset_items_message(self) -> str: - head = "App config items are not set in app config:\n" - return head + "\n".join([item.name for item in self.unset_items]) - - def get_user_message(self): - m = [] - if len(self.error_entries) > 0: - m.append(super().get_user_message()) - if len(self.unset_items) > 0: - m.append(self.unset_items_message) - return "\n".join(m) - - class AppConfigEntryValueFormatError(AppConfigFileEntryError): @property def friendly_message_head(self) -> str: @@ -162,11 +127,12 @@ class AppConfigItemNotSetError(AppConfigError): def __init__( self, message: str, + configuration: Configuration, items: list[ConfigItem], *args, **kwargs, ) -> None: - super().__init__(message, *args, **kwargs) + super().__init__(message, configuration, *args, **kwargs) self._items = items @@ -258,8 +224,45 @@ class ConfigManager(AppCommandFeatureProvider): def config_file_path(self) -> AppFeaturePath: return self._config_file_path - def get_config_str_dict(self) -> dict[str, str]: + @property + def all_set(self) -> bool: + return self.configuration.all_set + + def get_item(self, name: str) -> ConfigItem[Any]: + if not name.startswith(self.config_name_prefix + "_"): + name = f"{self.config_name_prefix}_{name}" + + item = self.configuration.get_or(name, None) + if item is None: + raise AppConfigError(f"Config item '{name}' not found.", self.configuration) + return item + + @overload + def get_item_value_str(self, name: str) -> str: ... + + @overload + def get_item_value_str(self, name: str, ensure_set: Literal[True]) -> str: ... + + @overload + def get_item_value_str(self, name: str, ensure_set: bool = True) -> str | None: ... + + def get_item_value_str(self, name: str, ensure_set: bool = True) -> str | None: self.reload_config_file() + item = self.get_item(name) + if ensure_set and not item.is_set: + raise AppConfigItemNotSetError( + f"Config item '{name}' is not set.", self.configuration, [item] + ) + return item.value_str + + def get_str_dict(self, ensure_all_set: bool = True) -> dict[str, str]: + self.reload_config_file() + if ensure_all_set and not self.configuration.all_set: + raise AppConfigItemNotSetError( + "Some config items are not set.", + self.configuration, + self.configuration.get_unset_items(), + ) return self.configuration.to_str_dict() def get_domain_item_name(self) -> str: @@ -343,26 +346,6 @@ class ConfigManager(AppCommandFeatureProvider): return entry_dict - def _check_key( - self, entry_dict: dict[str, SimpleLineConfigParser.Entry] - ) -> dict[str, SimpleLineConfigParser.Entry]: - undefined: list[SimpleLineConfigParser.Entry] = [] - for key, entry in entry_dict.items(): - if not self.configuration.has_key(key): - undefined.append(entry) - unset_items: list[ConfigItem] = [] - for item in self.configuration: - if item.name not in entry_dict or entry_dict[item.name].value == "": - unset_items.append(item) - if len(undefined) > 0 or len(unset_items) > 0: - raise AppConfigEntryKeyError( - "Entry keys are not defined in app config.", - self.configuration, - undefined, - unset_items, - ) - return entry_dict - def _check_type( self, entry_dict: dict[str, SimpleLineConfigParser.Entry] ) -> dict[str, Any]: @@ -393,7 +376,6 @@ class ConfigManager(AppCommandFeatureProvider): parsed = self._parse_config_file() entry_groups = parsed.cru_iter().group_by(lambda e: e.key) entry_dict = self._check_duplicate(entry_groups) - entry_dict = self._check_key(entry_dict) value_dict = self._check_type(entry_dict) return value_dict @@ -401,6 +383,8 @@ class ConfigManager(AppCommandFeatureProvider): self.configuration.reset_all() value_dict = self._read_config_file() for key, value in value_dict.items(): + if value is None: + continue self.configuration.set_config_item(key, value) def _print_app_config_info(self): @@ -415,26 +399,22 @@ class ConfigManager(AppCommandFeatureProvider): dest="config_command", required=True, metavar="CONFIG_COMMAND" ) _init_parser = subparsers.add_parser( - "init", help="Create an initial configuration file." + "init", help="create an initial config file" ) _print_app_parser = subparsers.add_parser( "print-app", - help="Print application configuration information " - "of the items defined in the application.", - ) - _print_parser = subparsers.add_parser( - "print", help="Print current configuration." + help="print information of the config items defined by app", ) + _print_parser = subparsers.add_parser("print", help="print current config") _check_config_parser = subparsers.add_parser( "check", - help="Check the validity of the configuration file.", + help="check the validity of the config file", ) _check_config_parser.add_argument( "-f", "--format-only", action="store_true", - help="Only check content format, not " - "for application configuration requirements.", + help="only check content format, not app config item requirements.", ) def run_command(self, args) -> None: diff --git a/tools/cru-py/cru/service/_nginx.py b/tools/cru-py/cru/service/_nginx.py index a291866..ad29d21 100644 --- a/tools/cru-py/cru/service/_nginx.py +++ b/tools/cru-py/cru/service/_nginx.py @@ -1,31 +1,89 @@ from argparse import Namespace +import re from ._base import AppCommandFeatureProvider from ._config import ConfigManager +from ._template import TemplateManager class NginxManager(AppCommandFeatureProvider): - def __init__(self): + def __init__(self) -> None: super().__init__("nginx-manager") + self._domains_cache: list[str] | None = None + self._domain_config_value_cache: str | None = None def setup(self) -> None: pass @property - def _template_domain_variable(self) -> str: + def domains(self) -> list[str]: + if self._domains_cache is None: + self._domains_cache = self._get_domains() + return self._domains_cache + + @property + def _domain_config_name(self) -> str: return self.app.get_feature(ConfigManager).get_domain_item_name() - def _create_domain_regex(self): - raise NotImplementedError() + def _get_domain_config_value(self) -> str: + if self._domain_config_value_cache is None: + self._domain_config_value_cache = self.app.get_feature( + ConfigManager + ).get_item_value_str(self._domain_config_name) + return self._domain_config_value_cache + + def _get_domains_from_text(self, text: str) -> set[str]: + domains: set[str] = set() + regex = re.compile(r"server_name\s+(\S+)\s*;") + domain_variable_str = f"${self._domain_config_name}" + brace_domain_variable_regex = re.compile( + r"\$\{\s*" + self._domain_config_name + r"\s*\}" + ) + for match in regex.finditer(text): + domain_part = match.group(1) + if domain_variable_str in domain_part: + domains.add( + domain_part.replace( + domain_variable_str, self._get_domain_config_value() + ) + ) + continue + m = brace_domain_variable_regex.search(domain_part) + if m: + domains.add( + domain_part.replace(m.group(0), self._get_domain_config_value()) + ) + continue + domains.add(domain_part) + return domains + + def _get_nginx_conf_template_text(self) -> str: + template_manager = self.app.get_feature(TemplateManager) + text = "" + for path, template in template_manager.template_tree.templates: + if path.as_posix().startswith("nginx/"): + text += template.raw_text + return text def _get_domains(self) -> list[str]: - raise NotImplementedError() + text = self._get_nginx_conf_template_text() + domains = list(self._get_domains_from_text(text)) + domains.remove(self._get_domain_config_value()) + return [self._get_domain_config_value(), *domains] + + def _print_domains(self) -> None: + for domain in self.domains: + print(domain) def get_command_info(self): - raise NotImplementedError() + return "nginx", "Manage nginx related things." def setup_arg_parser(self, arg_parser): - raise NotImplementedError() + subparsers = arg_parser.add_subparsers( + dest="nginx_command", required=True, metavar="NGINX_COMMAND" + ) + _list_parser = subparsers.add_parser("list", help="list domains") def run_command(self, args: Namespace) -> None: - raise NotImplementedError() + if args.nginx_command == "list": + self._print_domains() diff --git a/tools/cru-py/cru/service/_template.py b/tools/cru-py/cru/service/_template.py index 7d479df..9241a1f 100644 --- a/tools/cru-py/cru/service/_template.py +++ b/tools/cru-py/cru/service/_template.py @@ -41,14 +41,14 @@ class TemplateManager(AppCommandFeatureProvider): ) return self._template_tree - def print_file_lists(self) -> None: + def _print_file_lists(self) -> None: for file in CruIterator(self.template_tree.templates).transform(lambda t: t[0]): print(file.as_posix()) - def generate_files(self) -> None: + def _generate_files(self) -> None: config_manager = self.app.get_feature(ConfigManager) self.template_tree.generate_to( - self.generated_dir.full_path_str, config_manager.get_config_str_dict() + self.generated_dir.full_path_str, config_manager.get_str_dict() ) def get_command_info(self): @@ -58,17 +58,17 @@ class TemplateManager(AppCommandFeatureProvider): subparsers = arg_parser.add_subparsers( dest="template_command", required=True, metavar="TEMPLATE_COMMAND" ) - _list_parser = subparsers.add_parser("list", help="List templates.") + _list_parser = subparsers.add_parser("list", help="list templates") _variables_parser = subparsers.add_parser( - "variables", help="List variables for a specific template." + "variables", help="list variables used in all templates" ) - _generate_parser = subparsers.add_parser("generate", help="Generate template.") + _generate_parser = subparsers.add_parser("generate", help="generate templates") def run_command(self, args: Namespace) -> None: if args.template_command == "list": - self.print_file_lists() + self._print_file_lists() elif args.template_command == "variables": for var in self.template_tree.variables: print(var) elif args.template_command == "generate": - self.generate_files() + self._generate_files() diff --git a/tools/cru-py/cru/template.py b/tools/cru-py/cru/template.py index eb8c8ab..2b0f1bc 100644 --- a/tools/cru-py/cru/template.py +++ b/tools/cru-py/cru/template.py @@ -29,6 +29,10 @@ class CruTemplate: return self._prefix @property + def raw_text(self) -> str: + return self._template.template + + @property def py_template(self) -> Template: return self._template |