From 0d7dd933dfab91faf1a42b7883eb059d7fb31d80 Mon Sep 17 00:00:00 2001
From: crupest <crupest@outlook.com>
Date: Mon, 11 Nov 2024 01:12:29 +0800
Subject: HALF WORK: 2024.1.11
---
tools/cru-py/cru/_iter.py | 29 ++++++++++++--
tools/cru-py/cru/config.py | 21 ++++++++--
tools/cru-py/cru/parsing.py | 78 ++++++++++++++++++++++---------------
tools/cru-py/cru/service/_config.py | 16 +++++++-
4 files changed, 103 insertions(+), 41 deletions(-)
(limited to 'tools/cru-py')
diff --git a/tools/cru-py/cru/_iter.py b/tools/cru-py/cru/_iter.py
index 5d3766a..8f58561 100644
--- a/tools/cru-py/cru/_iter.py
+++ b/tools/cru-py/cru/_iter.py
@@ -63,7 +63,7 @@ class _Generic:
@staticmethod
def aggregate(
- *results: _Generic.StepAction[_V, _R]
+ *results: _Generic.StepAction[_V, _R],
) -> _Generic.StepAction[_V, _R]:
return _Generic.StepAction(results, _Generic.StepActionKind.AGGREGATE)
@@ -255,7 +255,6 @@ class _Helpers:
class _Creators:
-
class Raw:
@staticmethod
def empty() -> Iterator[Never]:
@@ -313,7 +312,7 @@ class CruIterator(Generic[_T]):
@staticmethod
def _wrap(
- f: Callable[Concatenate[CruIterator[_T], _P], Iterable[_O]]
+ f: Callable[Concatenate[CruIterator[_T], _P], Iterable[_O]],
) -> Callable[Concatenate[CruIterator[_T], _P], CruIterator[_O]]:
def _wrapped(
self: CruIterator[_T], *args: _P.args, **kwargs: _P.kwargs
@@ -435,11 +434,33 @@ class CruIterator(Generic[_T]):
value_set = set(old_values)
return self.transform(lambda v: new_value if v in value_set else v)
+ def group_by(self, key_getter: Callable[[_T], _O]) -> dict[_O, list[_T]]:
+ result: dict[_O, list[_T]] = {}
+
+ for item in self:
+ key = key_getter(item)
+ if key not in result:
+ result[key] = []
+ result[key].append(item)
+
+ return result
+
+
+class CruIterMixin(Generic[_T]):
+ def cru_iter(self: Iterable[_T]) -> CruIterator[_T]:
+ return CruIterator(self)
+
+
+class CruIterList(list[_T], CruIterMixin[_T]):
+ pass
+
class CruIterable:
Generic: TypeAlias = _Generic
- Iterator: TypeAlias = CruIterator
+ Iterator: TypeAlias = CruIterator[_T]
Helpers: TypeAlias = _Helpers
+ Mixin: TypeAlias = CruIterMixin[_T]
+ IterList: TypeAlias = CruIterList[_T]
CRU.add_objects(CruIterable, CruIterator)
diff --git a/tools/cru-py/cru/config.py b/tools/cru-py/cru/config.py
index 926ed6a..497eb01 100644
--- a/tools/cru-py/cru/config.py
+++ b/tools/cru-py/cru/config.py
@@ -82,11 +82,20 @@ class ConfigItem(Generic[_T]):
def can_generate_default(self) -> bool:
return self.default is not None
- def set_value(self, v: _T | str, /, allow_convert_from_str=False):
- if allow_convert_from_str:
- self._value = self.value_type.check_value(v)
- else:
+ def set_value(
+ self, v: _T | str, *, empty_is_default=True, allow_convert_from_str=True
+ ):
+ if empty_is_default and v == "":
+ self._value = None
+ elif allow_convert_from_str:
self._value = self.value_type.check_value_or_try_convert_from_str(v)
+ else:
+ self._value = self.value_type.check_value(v)
+
+ def reset(self, clear_default_cache=False):
+ if clear_default_cache:
+ self._default_value = None
+ self._value = None
def generate_default_value(self) -> _T:
if self.default is None:
@@ -140,3 +149,7 @@ class Configuration(CruUniqueKeyList[ConfigItem, str]):
item = ConfigItem(name, description, INTEGER_VALUE_TYPE, value, default)
self.add(item)
return item
+
+ def reset_all(self, clear_default_cache=False) -> None:
+ for item in self:
+ item.reset(clear_default_cache)
diff --git a/tools/cru-py/cru/parsing.py b/tools/cru-py/cru/parsing.py
index a9eee04..5049a33 100644
--- a/tools/cru-py/cru/parsing.py
+++ b/tools/cru-py/cru/parsing.py
@@ -1,19 +1,33 @@
+from __future__ import annotations
+
from abc import ABCMeta, abstractmethod
-from typing import TypeVar, Generic, NoReturn, Callable
+from typing import NamedTuple, TypeAlias, TypeVar, Generic, NoReturn, Callable
from ._error import CruException
+from ._iter import CruIterable
_T = TypeVar("_T")
-class ParseException(CruException):
+class ParseException(CruException, Generic[_T]):
def __init__(
- self, message, text: str, line_number: int | None = None, *args, **kwargs
+ self,
+ message,
+ parser: Parser[_T],
+ text: str,
+ line_number: int | None = None,
+ *args,
+ **kwargs,
):
super().__init__(message, *args, **kwargs)
+ self._parser = parser
self._text = text
self._line_number = line_number
+ @property
+ def parser(self) -> Parser[_T]:
+ return self._parser
+
@property
def text(self) -> str:
return self._text
@@ -38,16 +52,34 @@ class Parser(Generic[_T], metaclass=ABCMeta):
def raise_parse_exception(
self, text: str, line_number: int | None = None
) -> NoReturn:
- a = f" at line {line_number}" if line_number is not None else ""
- raise ParseException(f"Parser {self.name} failed{a}.", text, line_number)
+ a = line_number and f" at line {line_number}" or ""
+ raise ParseException(f"Parser {self.name} failed{a}.", self, text, line_number)
+
+
+class SimpleLineConfigParserItem(NamedTuple):
+ key: str
+ value: str
+ line_number: int | None = None
-class SimpleLineConfigParser(Parser[list[tuple[str, str]]]):
+SimpleLineConfigParserResult: TypeAlias = CruIterable.IterList[
+ SimpleLineConfigParserItem
+]
+
+
+class SimpleLineConfigParser(Parser[SimpleLineConfigParserResult]):
+ """
+ The parsing result is a list of tuples (key, value, line number).
+ """
+
+ Item: TypeAlias = SimpleLineConfigParserItem
+ Result: TypeAlias = SimpleLineConfigParserResult
+
def __init__(self) -> None:
super().__init__(type(self).__name__)
- def _parse(self, s: str, callback: Callable[[str, str], None]) -> None:
- for ln, line in enumerate(s.splitlines()):
+ def _parse(self, text: str, callback: Callable[[Item], None]) -> None:
+ for ln, line in enumerate(text.splitlines()):
line_number = ln + 1
# check if it's a comment
if line.strip().startswith("#"):
@@ -59,27 +91,9 @@ class SimpleLineConfigParser(Parser[list[tuple[str, str]]]):
key, value = line.split("=", 1)
key = key.strip()
value = value.strip()
- callback(key, value)
-
- def parse(self, s: str) -> list[tuple[str, str]]:
- items = []
- self._parse(s, lambda key, value: items.append((key, value)))
- return items
-
- def parse_to_dict(
- self, s: str, /, allow_override: bool = False
- ) -> tuple[dict[str, str], list[tuple[str, str]]]:
- result: dict[str, str] = {}
- duplicate: list[tuple[str, str]] = []
-
- def add(key: str, value: str) -> None:
- if key in result:
- if allow_override:
- duplicate.append((key, result[key]))
- result[key] = value
- else:
- self.raise_parse_exception(f"Key '{key}' already exists!", None)
- result[key] = value
-
- self._parse(s, add)
- return result, duplicate
+ callback(SimpleLineConfigParserItem(key, value, line_number))
+
+ def parse(self, text: str) -> Result:
+ result = SimpleLineConfigParserResult()
+ self._parse(text, lambda item: result.append(item))
+ return result
diff --git a/tools/cru-py/cru/service/_config.py b/tools/cru-py/cru/service/_config.py
index b5f3e7c..018b45b 100644
--- a/tools/cru-py/cru/service/_config.py
+++ b/tools/cru-py/cru/service/_config.py
@@ -5,6 +5,7 @@ from cru.value import (
RandomStringValueGenerator,
UuidValueGenerator,
)
+from cru.parsing import SimpleLineConfigParser
from ._base import AppFeaturePath, AppFeatureProvider, OWNER_NAME
@@ -83,13 +84,26 @@ class ConfigManager(AppFeatureProvider):
)
@property
- def config_path(self) -> AppFeaturePath:
+ def config_file_path(self) -> AppFeaturePath:
return self._config_path
@property
def configuration(self) -> Configuration:
return self._configuration
+ @property
+ def config_keys(self) -> list[str]:
+ return [item.name for item in self.configuration]
+
@property
def config_map(self) -> dict[str, str]:
raise NotImplementedError()
+
+ def reload_config_file(self) -> bool:
+ self.configuration.reset_all()
+ if not self.config_file_path.check_self():
+ return False
+ parser = SimpleLineConfigParser()
+ parse_result = parser.parse(self.config_file_path.full_path.read_text())
+ config_dict = parse_result.cru_iter().group_by(lambda i: i.key)
+ return True
--
cgit v1.2.3