diff options
author | crupest <crupest@outlook.com> | 2024-11-11 01:12:29 +0800 |
---|---|---|
committer | Yuqian Yang <crupest@crupest.life> | 2024-12-18 18:31:27 +0800 |
commit | eff33fcbc8e78b1cd15332c229cd39ae9befbe5e (patch) | |
tree | 1c83d262c79a15a06b5a02ce3276af0e8130c513 /tools/cru-py/cru | |
parent | f35afa17a47fa8e0cafbff5956feab64adb503be (diff) | |
download | crupest-eff33fcbc8e78b1cd15332c229cd39ae9befbe5e.tar.gz crupest-eff33fcbc8e78b1cd15332c229cd39ae9befbe5e.tar.bz2 crupest-eff33fcbc8e78b1cd15332c229cd39ae9befbe5e.zip |
HALF WORK: 2024.11.11
Diffstat (limited to 'tools/cru-py/cru')
-rw-r--r-- | tools/cru-py/cru/attr.py | 287 | ||||
-rw-r--r-- | tools/cru-py/cru/excp.py | 93 | ||||
-rw-r--r-- | tools/cru-py/cru/paths.py | 2 | ||||
-rw-r--r-- | tools/cru-py/cru/service/nginx.py | 9 | ||||
-rw-r--r-- | tools/cru-py/cru/system.py | 2 | ||||
-rw-r--r-- | tools/cru-py/cru/util/__init__.py | 16 | ||||
-rw-r--r-- | tools/cru-py/cru/util/_const.py | 51 | ||||
-rw-r--r-- | tools/cru-py/cru/util/_func.py | 46 | ||||
-rw-r--r-- | tools/cru-py/cru/util/_list.py | 272 | ||||
-rw-r--r-- | tools/cru-py/cru/value.py | 10 |
10 files changed, 505 insertions, 283 deletions
diff --git a/tools/cru-py/cru/attr.py b/tools/cru-py/cru/attr.py index 5d4f276..32cca8d 100644 --- a/tools/cru-py/cru/attr.py +++ b/tools/cru-py/cru/attr.py @@ -1,48 +1,40 @@ import copy -from collections.abc import Callable +from collections.abc import Callable, Iterable from dataclasses import dataclass, field -from typing import Any +from typing import Any, ClassVar -from .util import CanBeList, TypeSet, L, ListOperations +from .util import CanBeList, TypeSet, F, L, WF, CruUniqueKeyInplaceList, CRU_NOT_FOUND, CRU_USE_DEFAULT, \ + CRU_DONT_CHANGE, CRU_PLACEHOLDER @dataclass class CruAttr: + USE_DEFAULT: ClassVar = CRU_USE_DEFAULT + name: str value: Any description: str - -class _CruAttrVoid: - _i = False - - def __init__(self): - if self._i: - raise ValueError("CruAttrNotSet is a singleton!") - self._i = True - - def __copy__(self): - return self - - -CRU_ATTR_NOT_SET = _CruAttrVoid() -CRU_ATTR_USE_DEFAULT = CRU_ATTR_NOT_SET.__copy__() + @staticmethod + def make(name: str, value: Any = USE_DEFAULT, description: str | None = None) -> "CruAttr": + return CruAttr(name, value, description) -def _is_attr_void(v: Any) -> bool: - return isinstance(v, _CruAttrVoid) +CruAttrDefaultFactory = Callable[["CruAttrDef"], Any] +CruAttrTransformer = Callable[[Any, "CruAttrDef"], Any] +CruAttrValidator = Callable[[Any, "CruAttrDef"], None] @dataclass class CruAttrDef: name: str description: str - default_factory: Callable[[], Any] - transformer: Callable[[Any], Any] - validator: Callable[[Any], None] + default_factory: CruAttrDefaultFactory + transformer: CruAttrTransformer + validator: CruAttrValidator - def __init__(self, name: str, description: str, default_factory: Callable[[], Any], - transformer: Callable[[Any], Any], validator: Callable[[Any], None]) -> None: + def __init__(self, name: str, description: str, default_factory: CruAttrDefaultFactory, + transformer: CruAttrTransformer, validator: CruAttrValidator) -> None: self.name = name self.description = description self.default_factory = default_factory @@ -51,22 +43,22 @@ class CruAttrDef: def transform(self, value: Any) -> Any: if self.transformer is not None: - return self.transformer(value) + return self.transformer(value, self) return value def validate(self, value: Any, /, force_allow_none: bool = False) -> None: if force_allow_none is value is None: return if self.validator is not None: - self.validator(value) + self.validator(value, self) - def transform_and_validate(self, value: Any = CRU_ATTR_USE_DEFAULT, /, force_allow_none: bool = False) -> Any: + def transform_and_validate(self, value: Any, /, force_allow_none: bool = False) -> Any: value = self.transform(value) self.validate(value, force_allow_none) return value def make_default_value(self) -> Any: - return self.transform_and_validate(self.default_factory()) + return self.transform_and_validate(self.default_factory(self)) def adopt(self, attr: CruAttr) -> CruAttr: attr = copy.deepcopy(attr) @@ -76,7 +68,7 @@ class CruAttrDef: elif attr.name != self.name: raise ValueError(f"Attr name is not match: {attr.name} != {self.name}") - if attr.value is CRU_ATTR_NOT_SET: + if attr.value is CruAttr.USE_DEFAULT: attr.value = self.make_default_value() else: attr.value = self.transform_and_validate(attr.value) @@ -86,9 +78,8 @@ class CruAttrDef: return attr - def make(self, value: Any = CRU_ATTR_USE_DEFAULT, description: None | str = None) -> CruAttr: - value = self.default_factory() if _is_attr_void(value) else value - value = self.default_factory() if value is CRU_ATTR_USE_DEFAULT else value + def make(self, value: Any = CruAttr.USE_DEFAULT, description: None | str = None) -> CruAttr: + value = self.make_default_value() if value is CruAttr.USE_DEFAULT else value value = self.transform_and_validate(value) return CruAttr(self.name, value, description if description is not None else self.description) @@ -96,17 +87,19 @@ class CruAttrDef: @dataclass class CruAttrDefBuilder: + DONT_CHANGE: ClassVar = CRU_DONT_CHANGE + name: str description: str types: CanBeList[type] = field(default=None) allow_none: bool = field(default=True) - default: Any = field(default=CRU_ATTR_NOT_SET) - default_factory: Callable[[], Any] | None = field(default=None) + default: Any = field(default=CruAttr.USE_DEFAULT) + default_factory: CruAttrDefaultFactory | None = field(default=None) auto_list: bool = field(default=False) - transformers: list[Callable[[Any], Any]] = field(default_factory=list) - validators: list[Callable[[Any], None]] = field(default_factory=list) - override_transformer: Callable[[Any], Any] | None = field(default=None) - override_validator: Callable[[Any], None] | None = field(default=None) + transformers: list[CruAttrTransformer] = field(default_factory=list) + validators: list[CruAttrValidator] = field(default_factory=list) + override_transformer: CruAttrTransformer | None = field(default=None) + override_validator: CruAttrValidator | None = field(default=None) build_hook: Callable[[CruAttrDef], None] | None = field(default=None) @@ -116,12 +109,12 @@ class CruAttrDefBuilder: self.description = description def auto_adjust_default(self) -> "CruAttrDefBuilder": - if self.default is not CRU_ATTR_NOT_SET and self.default is not None: + if self.default is not CruAttr.USE_DEFAULT and self.default is not None: return self - if self.allow_none and _is_attr_void(self.default): + if self.allow_none and self.default is CruAttr.USE_DEFAULT: self.default = None if not self.allow_none and self.default is None: - self.default = CRU_ATTR_NOT_SET + self.default = CruAttr.USE_DEFAULT if self.auto_list and not self.allow_none: self.default = [] @@ -133,49 +126,63 @@ class CruAttrDefBuilder: self.description = default_description return self - def with_default(self, default: Any, is_factory: bool = False, - reserve_if_not_set: bool = False) -> "CruAttrDefBuilder": - if is_factory: - self.default_factory = default - else: - if _is_attr_void(default) and reserve_if_not_set: - return self - else: - self.default = default + def with_default(self, /, default: Any | DONT_CHANGE = DONT_CHANGE, + default_factory: CruAttrDefaultFactory | DONT_CHANGE = DONT_CHANGE) -> "CruAttrDefBuilder": + if default is not CruAttrDefBuilder.DONT_CHANGE: + self.default = default + if default_factory is not CruAttrDefBuilder.DONT_CHANGE: + self.default_factory = default_factory + return self + + def with_default_value(self, default: Any) -> "CruAttrDefBuilder": + self.default = default return self - def with_default_factory(self, default_factory: Callable[[], Any]) -> "CruAttrDefBuilder": - return self.with_default(default_factory, is_factory=True) + def with_default_factory(self, default_factory: CruAttrDefaultFactory) -> "CruAttrDefBuilder": + self.default_factory = default_factory + return self - def with_types(self, allowed_types: CanBeList[type], default: Any = CRU_ATTR_NOT_SET) -> "CruAttrDefBuilder": + def with_types(self, allowed_types: CanBeList[type], default: Any = DONT_CHANGE) -> "CruAttrDefBuilder": self.types = allowed_types - if _is_attr_void(default): + if default is not CruAttrDefBuilder.DONT_CHANGE: self.default = default return self - def with_allow_none(self, allow_none: bool, default: Any = CRU_ATTR_NOT_SET) -> "CruAttrDefBuilder": + def with_allow_none(self, allow_none: bool, default: Any = DONT_CHANGE) -> "CruAttrDefBuilder": self.allow_none = allow_none - if _is_attr_void(default): + if default is not CruAttrDefBuilder.DONT_CHANGE: self.default = default return self - def with_optional(self, default: Any = CRU_ATTR_NOT_SET) -> "CruAttrDefBuilder": + def with_optional(self, default: Any = DONT_CHANGE) -> "CruAttrDefBuilder": return self.with_allow_none(True, default) - def with_required(self, default: Any = CRU_ATTR_NOT_SET) -> "CruAttrDefBuilder": + def with_required(self, default: Any = DONT_CHANGE) -> "CruAttrDefBuilder": return self.with_allow_none(False, default) - def with_constraint(self, required: bool, allowed_types: CanBeList[type] = None, /, - default: Any = CRU_ATTR_NOT_SET, auto_list=False, *, default_is_factory: bool = False, - default_reserve_if_not_set: bool = True) -> "CruAttrDefBuilder": - return (self.with_allow_none(required).with_types(allowed_types).with_auto_list(auto_list). - with_default(default, default_is_factory, default_reserve_if_not_set)) + def with_constraint(self, /, required: bool = DONT_CHANGE, allowed_types: CanBeList[type] = DONT_CHANGE, + default: Any = DONT_CHANGE, default_factory: CruAttrDefaultFactory = DONT_CHANGE, + auto_list: bool = DONT_CHANGE) -> "CruAttrDefBuilder": + def should_change(v): + return v is not CruAttrDefBuilder.DONT_CHANGE + + if should_change(required): + self.allow_none = not required + if should_change(allowed_types): + self.types = allowed_types + if should_change(default): + self.default = default + if should_change(default_factory): + self.default_factory = default_factory + if should_change(auto_list): + self.auto_list = auto_list + return self def with_auto_list(self, transform_auto_list: bool = True) -> "CruAttrDefBuilder": self.auto_list = transform_auto_list return self - def add_transformer(self, transformer: Callable[[Any], Any] | None) -> "CruAttrDefBuilder": + def add_transformer(self, transformer: Callable[[Any, "CruAttrDef"], Any] | None) -> "CruAttrDefBuilder": if transformer is not None: self.transformers.append(transformer) return self @@ -184,7 +191,7 @@ class CruAttrDefBuilder: self.transformers.clear() return self - def add_validator(self, validator: Callable[[Any], None] | None) -> "CruAttrDefBuilder": + def add_validator(self, validator: Callable[[Any, "CruAttrDef"], None] | None) -> "CruAttrDefBuilder": if validator is not None: self.validators.append(validator) return self @@ -217,32 +224,38 @@ class CruAttrDefBuilder: valid, err = b.is_valid() if not valid: raise ValueError(err) - def composed_transformer(v: Any): + def composed_transformer(v: Any, d_): + transformers = L(b.transformers) + transformers.transform(lambda f: F(f).bind(CRU_PLACEHOLDER, d_)) + transformer = F.make_chain(*transformers) if not transformers.is_empty else WF.only_you if b.auto_list: v = L.make(v) - for t in b.transformers: - v = t(v) + v = v.transform(transformer) + else: + v = transformer(v) return v type_set = TypeSet(b.types) - def composed_validator(v: Any): + def composed_validator(v: Any, d_): + validators = L(b.validators) + validators.transform(lambda f: F(f).bind(CRU_PLACEHOLDER, d_)) + validator = F.make_chain(*validators) if not validators.is_empty else WF.true if b.auto_list: type_set.check_value_list(v, allow_none=b.allow_none) - ListOperations.foreach(v, *b.validators) + L(v).foreach(validator) else: type_set.check_value(v, allow_none=b.allow_none) - for vl in b.validators: vl(v) + validator(v) - validator = b.override_validator or composed_validator - transformer = b.override_transformer or composed_transformer + real_transformer = b.override_transformer or composed_transformer + real_validator = b.override_validator or composed_validator default_factory = b.default_factory - if b.default_factory is None: - default_factory = lambda: copy.deepcopy(b.default) - validator(default_factory()) + if default_factory is None: + default_factory = lambda _d: copy.deepcopy(b.default) - d = CruAttrDef(b.name, b.description, default_factory, transformer, validator) + d = CruAttrDef(b.name, b.description, default_factory, real_transformer, real_validator) if b.build_hook: b.build_hook(d) return d @@ -252,93 +265,77 @@ class CruAttrDefBuilder: return CruAttrDefBuilder._build(c, auto_adjust_default) -class CruAttrDefRegistry: +class CruAttrDefRegistry(CruUniqueKeyInplaceList[CruAttrDef, str]): def __init__(self) -> None: - self._def_list = [] - - @property - def items(self) -> list[CruAttrDef]: - return self._def_list - - def register(self, def_: CruAttrDef): - for i in self._def_list: - if i.name == def_.name: - raise ValueError(f"Attribute {def_.name} already exists!") - self._def_list.append(def_) + super().__init__(lambda d: d.name) def make_builder(self, name: str, default_description: str) -> CruAttrDefBuilder: b = CruAttrDefBuilder(name, default_description) - b.build_hook = lambda a: self.register(a) + b.build_hook = lambda a: self.add(a) return b - def get_item_optional(self, name: str) -> CruAttrDef | None: - for i in self._def_list: - if i.name == name: - return i - return None + def adopt(self, attr: CruAttr) -> CruAttr: + d = self.get(attr.name) + return d.adopt(attr) -class CruAttrTable: +class CruAttrTable(CruUniqueKeyInplaceList[CruAttr, str]): def __init__(self, registry: CruAttrDefRegistry) -> None: self._registry: CruAttrDefRegistry = registry - self._attrs: list[CruAttr] = [] - - def get_attr_or(self, name: str, fallback: Any = None) -> CruAttr | Any: - for a in self._attrs: - if a.name == name: - return a - return fallback - - def get_attr(self, name) -> CruAttr: - a = self.get_attr_or(name, None) - if a is None: - raise KeyError(f"Attribute {name} not found!") - return a - - def get_value_or(self, name: str, fallback: Any = None) -> Any: - a = self.get_attr_or(name, None) - if a is None: + super().__init__(lambda a: a.name, before_add=registry.adopt) + + @property + def registry(self) -> CruAttrDefRegistry: + return self._registry + + def get_value_or(self, name: str, fallback: Any = CRU_NOT_FOUND) -> Any: + a = self.get_or(name, CRU_NOT_FOUND) + if a is CRU_NOT_FOUND: return fallback return a.value def get_value(self, name: str) -> Any: - a = self.get_attr(name) + a = self.get(name) return a.value - def get_def_or(self, name: str, fallback: Any = None) -> CruAttrDef | Any: - d = self._registry.get_item_optional(name) - return d if d is not None else fallback + def make_attr(self, name: str, value: Any = CruAttr.USE_DEFAULT, /, description: str | None = None) -> CruAttr: + d = self._registry.get(name) + return d.make(value, description or d.description) - def get_def(self, name: str) -> CruAttrDef: - d = self._registry.get_item_optional(name) - if d is None: - raise KeyError(f"Attribute definition {name} not found!") - return d + def add_value(self, name: str, value: Any = CruAttr.USE_DEFAULT, /, description: str | None = None, *, + replace: bool = False) -> CruAttr: + attr = self.make_attr(name, value, description) + self.add(attr, replace) + return attr - def _check_already_exist(self, name: str) -> None: - if name in self: - raise KeyError(f"Attribute {name} already exists!") + def extend_values(self, *t: tuple[str, Any, str | None], replace: bool = False) -> None: + t = [self.make_attr(n, v, d) for n, v, d in t] + self.extend(t, replace) - def add_attr(self, attr: CruAttr) -> CruAttr: - self._check_already_exist(attr.name) - d = self.get_def(attr.name) - attr = d.adopt(attr) - self._attrs.append(attr) - return attr + def extend_with(self, a: CruAttr | Iterable[CruAttr | tuple[str, Any, str | None]] | dict[ + str, Any | tuple[Any, str]] | None = None, replace: bool = False): + if a is None: return - def add(self, name: str, value: Any = CRU_ATTR_USE_DEFAULT, /, description: str | None = None) -> CruAttr: - self._check_already_exist(name) - d = self.get_def(name) - attr = d.make(value, description or d.description) - self._attrs.append(attr) - return attr + if isinstance(a, CruAttr): + self.add(a, replace) + return + + if isinstance(a, dict): + l = L() + for k, v in a.items(): + if isinstance(v, tuple): + v, d = v + else: + d = None + l.append(self.make_attr(k, v, d)) + self.extend(l, replace) + return - def remove(self, name: str, /, allow_absense: bool = True) -> None: - l = [a for a in self._attrs if a.name == name] - if len(l) != len(self._attrs) and not allow_absense: - raise KeyError(f"Attribute {name} not found!") - self._attrs = l + if isinstance(a, Iterable): + l = L(a) + l.transform_if(lambda n_, v_, d_: self.make_attr(n_, v_, d_), F(isinstance).bind(CRU_PLACEHOLDER, tuple)) + self.extend(l, replace) + return - def __contains__(self, name): - return any(a.name == name for a in self._attrs) + raise TypeError(f"Unsupported type: {type(a)}") diff --git a/tools/cru-py/cru/excp.py b/tools/cru-py/cru/excp.py index 6734140..358ad90 100644 --- a/tools/cru-py/cru/excp.py +++ b/tools/cru-py/cru/excp.py @@ -1,13 +1,14 @@ from typing import Any -from .attr import CruAttrDefRegistry, CRU_ATTR_USE_DEFAULT, CruAttr, CruAttrTable -from .util import CanBeList +from .attr import CruAttrDefRegistry, CruAttr, CruAttrTable +from .util import CRU_NOT_FOUND, CruList, CRU_USE_DEFAULT CRU_EXCEPTION_ATTR_DEF_REGISTRY = CruAttrDefRegistry() -_REGISTRY = CRU_EXCEPTION_ATTR_DEF_REGISTRY -class CruException(Exception, CruAttrTable): +class CruException(Exception): + ATTR_REGISTRY = CRU_EXCEPTION_ATTR_DEF_REGISTRY + MESSAGE_KEY = "message" INNER_KEY = "inner" INTERNAL_KEY = "internal" @@ -16,82 +17,81 @@ class CruException(Exception, CruAttrTable): PATH_KEY = "path" TYPE_KEY = "type_" - _REGISTRY.make_builder(MESSAGE_KEY, "Message describing the exception.").with_constraint(True, str, - "").build() - _REGISTRY.make_builder("inner", "Inner exception.").with_constraint(True, Exception, - auto_list=True).build() - _REGISTRY.make_builder("internal", - "True if the exception is caused by wrong internal logic. False if it is caused by user's wrong input.").with_constraint( + ATTR_REGISTRY.make_builder(MESSAGE_KEY, "Message describing the exception.").with_constraint(True, str, + "").build() + ATTR_REGISTRY.make_builder(INNER_KEY, "Inner exception.").with_constraint(True, Exception, + auto_list=True).build() + ATTR_REGISTRY.make_builder(INTERNAL_KEY, + "True if the exception is caused by wrong internal logic. False if it is caused by user's wrong input.").with_constraint( True, bool, False).build() - _REGISTRY.make_builder(NAME_KEY, "Name of the object that causes the exception.").with_types(str).build() - _REGISTRY.make_builder(VALUE_KEY, "Value that causes the exception.").build() - _REGISTRY.make_builder(PATH_KEY, "Path that causes the exception.").with_types(str).build() - _REGISTRY.make_builder(TYPE_KEY, "Python type related to the exception.").with_types(type).build() + ATTR_REGISTRY.make_builder(NAME_KEY, "Name of the object that causes the exception.").with_types(str).build() + ATTR_REGISTRY.make_builder(VALUE_KEY, "Value that causes the exception.").build() + ATTR_REGISTRY.make_builder(PATH_KEY, "Path that causes the exception.").with_types(str).build() + ATTR_REGISTRY.make_builder(TYPE_KEY, "Python type related to the exception.").with_types(type).build() - # TODO: CONTINUE HERE TOMORROW! def __init__(self, message: str, *args, - init_attrs: dict[str, Any] | None = None, - attrs: dict[str, Any] | None = None, **kwargs) -> None: + init_attrs: list[CruAttr] | dict[str, Any] | None = None, + attrs: list[CruAttr] | dict[str, Any] | None = None, **kwargs) -> None: super().__init__(message, *args) - self._attrs = { - CruException.MESSAGE_KEY: message, - } + self._attrs: CruAttrTable = CruAttrTable(self.ATTR_REGISTRY) + + self._attrs.add_value(CruException.MESSAGE_KEY, message) if init_attrs is not None: - self._attrs.update(init_attrs) + self._attrs.extend_with(init_attrs, True) if attrs is not None: - self._attrs.update(attrs) - self._attrs.update(kwargs) + self._attrs.extend_with(attrs, True) + self._attrs.extend_with(dict(kwargs), True) + + @property + def attrs(self) -> CruAttrTable: + return self._attrs + + def get_attr(self, name: str) -> Any: + return self._attrs.get_value_or(name, None) @property def message(self) -> str: - return self[CruException.MESSAGE_KEY] + return self.get_attr(CruException.MESSAGE_KEY) @property def internal(self) -> bool: - return self[CruException.INTERNAL_KEY] + return self.get_attr(CruException.INTERNAL_KEY) @property def inner(self) -> list[Exception]: - return self[CruException.INNER_KEY] + return self.get_attr(CruException.INNER_KEY) @property def name(self) -> str | None: - return self[CruException.NAME_KEY] + return self.get_attr(CruException.NAME_KEY) @property def value(self) -> Any | None: - return self[CruException.VALUE_KEY] + return self.get_attr(CruException.VALUE_KEY) @property def path(self) -> str | None: - return self[CruException.PATH_KEY] + return self.get_attr(CruException.PATH_KEY) @property - def type(self) -> type | None: - return self[CruException.TYPE_KEY] + def type_(self) -> type | None: + return self.get_attr(CruException.TYPE_KEY) def _get_attr_list_recursive(self, name: str, depth: int, max_depth: int, l: list[Any]): if 0 < max_depth < depth + 1: return - a = self._attrs.get(name, None) - if a is not None: + a = self._attrs.get_or(name) + if a is not CRU_NOT_FOUND: l.append(a) for i in self.inner: if isinstance(i, CruException): i._get_attr_list_recursive(name, depth + 1, max_depth, l) - def get_attr_list_recursive(self, name: str, /, max_depth: int = -1) -> list[Any]: + def get_attr_list_recursive(self, name: str, /, max_depth: int = -1, des: type = CruList) -> list[Any]: l = [] self._get_attr_list_recursive(name, 0, max_depth, l) - return l - - def get_optional_attr(self, name: str, max_depth: int = -1) -> Any | None: - l = self.get_attr_list_recursive(name, max_depth) - return l[0] if len(l) > 0 else None - - def __getitem__(self, name: str) -> Any | None: - return self.get_optional_attr(name) + return des(l) class CruInternalLogicError(CruException): @@ -102,14 +102,15 @@ class CruInternalLogicError(CruException): class UserFriendlyException(CruException): USER_MESSAGE_KEY = "user_message" - CRU_EXCEPTION_ATTR_DEF_REGISTRY.register( - CruExceptionAttrDef(USER_MESSAGE_KEY, "Message describing the exception, but with user-friendly language.")) + CruException.ATTR_REGISTRY.make_builder(USER_MESSAGE_KEY, + "Message describing the exception, but with user-friendly language.").with_types( + str).build() - def __init__(self, message: str, user_message: str | None = None, *args, **kwargs) -> None: + def __init__(self, message: str, user_message: str | None = CRU_USE_DEFAULT, *args, **kwargs) -> None: if user_message is None: user_message = message super().__init__(message, *args, init_attrs={UserFriendlyException.USER_MESSAGE_KEY: user_message}, **kwargs) @property def user_message(self) -> str: - return self[UserFriendlyException.USER_MESSAGE_KEY] + return self.get_attr(UserFriendlyException.USER_MESSAGE_KEY) diff --git a/tools/cru-py/cru/paths.py b/tools/cru-py/cru/paths.py index df5042b..937d2ac 100644 --- a/tools/cru-py/cru/paths.py +++ b/tools/cru-py/cru/paths.py @@ -1,5 +1,5 @@ -from pathlib import Path import os +from pathlib import Path from .excp import CruException diff --git a/tools/cru-py/cru/service/nginx.py b/tools/cru-py/cru/service/nginx.py index 9298623..f9120de 100644 --- a/tools/cru-py/cru/service/nginx.py +++ b/tools/cru-py/cru/service/nginx.py @@ -1,9 +1,10 @@ -from typing import Literal, Any, cast, ClassVar +import json import os -from os.path import join, basename, dirname -import subprocess import re -import json +import subprocess +from os.path import join, basename, dirname +from typing import Literal, Any, cast, ClassVar + import jsonschema from crupest.template2 import Template2 diff --git a/tools/cru-py/cru/system.py b/tools/cru-py/cru/system.py index 2e05cd1..4c9de01 100644 --- a/tools/cru-py/cru/system.py +++ b/tools/cru-py/cru/system.py @@ -1,5 +1,5 @@ -import re import os.path +import re def check_debian_derivative_version(name: str) -> None | str: diff --git a/tools/cru-py/cru/util/__init__.py b/tools/cru-py/cru/util/__init__.py index 83ab9f3..2cbb8f4 100644 --- a/tools/cru-py/cru/util/__init__.py +++ b/tools/cru-py/cru/util/__init__.py @@ -1,5 +1,8 @@ -from ._func import CruFunction, MetaFunction, RawFunctions, WrappedFunctions, PLACEHOLDER -from ._list import CruList, ListOperations, CanBeList +from ._const import make_unique_object, make_bool_unique_object, CRU_NOT_FOUND, CRU_USE_DEFAULT, CRU_DONT_CHANGE, \ + CRU_PLACEHOLDER +from ._func import CruFunction, MetaFunction, RawFunctions, WrappedFunctions +from ._list import CruList, CruInplaceList, CruUniqueKeyInplaceList, ListOperations, CanBeList, ElementOperation, \ + ElementPredicate, ElementTransformer, OptionalElementOperation, ElementPredicate, OptionalElementTransformer from ._type import TypeSet F = CruFunction @@ -7,8 +10,11 @@ WF = WrappedFunctions L = CruList __all__ = [ - "CruFunction", "MetaFunction", "RawFunctions", "WrappedFunctions", "PLACEHOLDER", - "CruList", "ListOperations", "CanBeList", + "CRU_NOT_FOUND", "CRU_USE_DEFAULT", "CRU_DONT_CHANGE", "CRU_PLACEHOLDER", + "CruFunction", "MetaFunction", "RawFunctions", "WrappedFunctions", + "CruList", "CruInplaceList", "CruUniqueKeyInplaceList", "ListOperations", + "CanBeList", "ElementOperation", "ElementPredicate", "ElementTransformer", + "OptionalElementOperation", "ElementPredicate", "OptionalElementTransformer", "TypeSet", - "F", "WF", "L", + "F", "WF", "L" ] diff --git a/tools/cru-py/cru/util/_const.py b/tools/cru-py/cru/util/_const.py new file mode 100644 index 0000000..ea67450 --- /dev/null +++ b/tools/cru-py/cru/util/_const.py @@ -0,0 +1,51 @@ +from typing import Any + + +def make_unique_object() -> Any: + class _CruUnique: + _i = False + + def __init__(self): + if self._i: + raise ValueError("_CruAttrNotSet is a singleton!") + self._i = True + + def __copy__(self): + return self + + def __eq__(self, other): + return isinstance(other, _CruUnique) + + v = _CruUnique() + + return v + + +def make_bool_unique_object(b: bool) -> Any: + class _CruBoolUnique: + _i = False + + def __init__(self): + super().__init__(b) + if self._i: + raise ValueError("_CruAttrNotSet is a singleton!") + self._i = True + + def __copy__(self): + return self + + def __eq__(self, other): + return isinstance(other, _CruBoolUnique) or b == other + + def __bool__(self): + return b + + v = _CruBoolUnique() + + return v + + +CRU_NOT_FOUND = make_bool_unique_object(False) +CRU_USE_DEFAULT = make_unique_object() +CRU_DONT_CHANGE = make_unique_object() +CRU_PLACEHOLDER = make_unique_object() diff --git a/tools/cru-py/cru/util/_func.py b/tools/cru-py/cru/util/_func.py index 5fb49a9..3221c94 100644 --- a/tools/cru-py/cru/util/_func.py +++ b/tools/cru-py/cru/util/_func.py @@ -1,7 +1,7 @@ -from collections.abc import Callable, Iterable +from collections.abc import Callable from typing import TypeVar, Any, ParamSpec -from ._list import ListOperations, CruList +from ._const import CRU_PLACEHOLDER T = TypeVar("T") R = TypeVar("R") @@ -10,35 +10,28 @@ P = ParamSpec("P") P1 = ParamSpec("P1") -class _Placeholder: - pass - - -PLACEHOLDER = _Placeholder() - - class RawFunctions: @staticmethod - def ignore(*_v, **_kwargs) -> None: + def none(*_v, **_kwargs) -> None: return None @staticmethod - def true_(*_v, **_kwargs) -> True: + def true(*_v, **_kwargs) -> True: return True @staticmethod - def false_(*_v, **_kwargs) -> False: + def false(*_v, **_kwargs) -> False: return False @staticmethod - def i_dont_care(r: T, *_v, **_kwargs) -> T: - return r - - @staticmethod def identity(v: T) -> T: return v @staticmethod + def only_you(r: T, *_v, **_kwargs) -> T: + return r + + @staticmethod def equal(a: Any, b: Any) -> bool: return a == b @@ -58,7 +51,7 @@ class MetaFunction: popped = 0 real_args = [] for a in bind_args: - if isinstance(a, _Placeholder): + if a is CRU_PLACEHOLDER: real_args.append(args[popped]) popped += 1 else: @@ -129,22 +122,17 @@ class CruFunction: def __call__(self, *args, **kwargs): return self._f(*args, **kwargs) - def list_transform(self, l: Iterable[T]) -> CruList[T]: - return CruList(l).transform(self) - - def list_all(self, l: Iterable[T]) -> bool: - return ListOperations.all(l, self) - - def list_any(self, l: Iterable[T]) -> bool: - return ListOperations.any(l, self) - - def list_remove_all_if(self, l: Iterable[T]) -> CruList[T]: - return CruList(ListOperations.remove_all_if(l, self)) + @staticmethod + def make_chain(*fs: Callable) -> Callable[P, R1]: + return CruFunction(MetaFunction.chain(*fs)) class WrappedFunctions: + none = CruFunction(RawFunctions.none) + true = CruFunction(RawFunctions.true) + false = CruFunction(RawFunctions.false) identity = CruFunction(RawFunctions.identity) - ignore = CruFunction(RawFunctions.ignore) + only_you = CruFunction(RawFunctions.only_you) equal = CruFunction(RawFunctions.equal) not_equal = CruFunction(RawFunctions.not_equal) not_ = CruFunction(RawFunctions.not_) diff --git a/tools/cru-py/cru/util/_list.py b/tools/cru-py/cru/util/_list.py index 3b4cb28..92cd88c 100644 --- a/tools/cru-py/cru/util/_list.py +++ b/tools/cru-py/cru/util/_list.py @@ -1,17 +1,32 @@ from collections.abc import Iterable, Callable from typing import TypeVar, ParamSpec, Any, Generic +from ._const import CRU_NOT_FOUND + T = TypeVar("T") R = TypeVar("R") P = ParamSpec("P") CanBeList = T | Iterable[T] | None +ElementOperation = Callable[[T], Any] +ElementPredicate = Callable[[T], bool] +ElementTransformer = Callable[[Any], Any] +OptionalElementOperation = Callable[[T], Any] | None +OptionalElementTransformer = Callable[[Any], Any] | None class ListOperations: @staticmethod - def foreach(l: Iterable[T], *f: Callable[[T], None] | None) -> None: + def sub_by_indices(l: Iterable[T], *index: int) -> list[T]: + return [v for i, v in enumerate(l) if i in index] + + @staticmethod + def complement_indices(length: int, *index: int) -> list[int]: + return [i for i in range(length) if i not in index] + + @staticmethod + def foreach(l: Iterable[T], *f: OptionalElementOperation[T]) -> None: if len(f) == 0: return for v in l: for f_ in f: @@ -19,88 +34,131 @@ class ListOperations: f_(v) @staticmethod - def transform(l: Iterable[T], *f: Callable | None) -> list[R]: - r = [] + def all(l: Iterable[T], p: ElementPredicate[T]) -> bool: for v in l: - for f_ in f: - if f_ is not None: - v = f_(v) - r.append(v) - return r + if not p(v): return False + return True @staticmethod - def transform_if(l: Iterable[T], f: Callable[[T], R] | None, p: Callable[[T], bool] | None) -> list[R]: - return [(f(v) if f else v) for v in l if (p(v) if p else True)] + def all_is_instance(l: Iterable[T], *t: type) -> bool: + return all(type(v) in t for v in l) @staticmethod - def all(l: Iterable[T], *p: Callable[[T], bool] | None) -> bool: - if len(p) == 0 or all(v is None for v in p): - raise ValueError("At least one predicate is required!") + def any(l: Iterable[T], p: ElementPredicate[T]) -> bool: for v in l: - for p_ in p: - if p_ is not None and not p_(v): return False - return True + if p(v): return True + return False @staticmethod - def all_is_instance(l: Iterable[T], *t: type) -> bool: - return all(type(v) in t for v in l) + def find_all_if(l: Iterable[T], p: ElementPredicate[T]) -> list[T]: + return [v for v in l if p(v)] + + @staticmethod + def find_if(l: Iterable[T], p: ElementPredicate[T]) -> T | CRU_NOT_FOUND: + r = ListOperations.find_all_if(l, p) + return r[0] if len(r) > 0 else CRU_NOT_FOUND + + @staticmethod + def find_all_indices_if(l: Iterable[T], p: ElementPredicate[T]) -> list[int]: + return [i for i, v in enumerate(l) if p(v)] @staticmethod - def any(l: Iterable[T], *p: Callable[[T], bool] | None) -> bool: - if len(p) == 0 or all(v is None for v in p): - raise ValueError("At least one predicate is required!") + def find_index_if(l: Iterable[T], p: ElementPredicate[T]) -> int | CRU_NOT_FOUND: + r = ListOperations.find_all_indices_if(l, p) + return r[0] if len(r) > 0 else CRU_NOT_FOUND + + @staticmethod + def transform(l: Iterable[T], *f: OptionalElementTransformer) -> list[Any]: + r = [] for v in l: - for p_ in p: - if p_ is not None and p_(v): return True - return False + for f_ in f: + if f_ is not None: + v = f_(v) + r.append(v) + return r @staticmethod - def make(v: CanBeList[T], /, none_to_empty_list: bool = True) -> list[T]: - if v is None and none_to_empty_list: return [] - return v if isinstance(v, Iterable) else [v] + def transform_if(l: Iterable[T], f: OptionalElementTransformer, p: ElementPredicate[T]) -> list[Any]: + return [(f(v) if f else v) for v in l if p(v)] @staticmethod - def remove_all_if(l: Iterable[T], *p: Callable[[], bool] | None) -> list[T]: - def stay(v): - for p_ in p: - if p_ is not None and p_(): return False - return True + def remove_by_indices(l: Iterable[T], *index: int | None) -> list[T]: + return [v for i, v in enumerate(l) if i not in index] - return [v for v in l if stay(v)] + @staticmethod + def remove_if(l: Iterable[T], p: ElementPredicate[T]) -> list[T]: + i = ListOperations.find_index_if(l, p) + return ListOperations.remove_by_indices(l, i) + + @staticmethod + def remove_all_if(l: Iterable[T], p: ElementPredicate[T]) -> list[T]: + return [v for v in l if not p(v)] @staticmethod - def remove_all_value(l: Iterable[T], *r) -> list[T]: + def remove_all_value(l: Iterable[T], *r: Any) -> list[T]: return [v for v in l if v not in r] @staticmethod def replace_all_value(l: Iterable[T], old_value: Any, new_value: R) -> list[T | R]: return [new_value if v == old_value else v for v in l] + @staticmethod + def make(v: CanBeList[T], /, none_to_empty_list: bool = True) -> list[T]: + if v is None and none_to_empty_list: return [] + return list(v) if isinstance(v, Iterable) else [v] + class CruList(list, Generic[T]): + @property + def is_empty(self) -> bool: + return len(self) == 0 - def foreach(self, *f: Callable[[T], None] | None) -> None: - ListOperations.foreach(self, *f) + def sub_by_indices(self, *index: int) -> "CruList"[T]: + return CruList(ListOperations.sub_by_indices(self, *index)) - def transform(self, *f: Callable[[T], R] | None) -> "CruList"[R]: - return CruList(ListOperations.transform(self, *f)) + def complement_indices(self, *index: int) -> list[int]: + return ListOperations.complement_indices(len(self), *index) - def transform_if(self, f: Callable[[T], R] | None, p: Callable[[T], bool] | None) -> "CruList"[R]: - return CruList(ListOperations.transform_if(self, f, p)) + def foreach(self, *f: OptionalElementOperation[T]) -> None: + ListOperations.foreach(self, *f) - def all(self, *p: Callable[[T], bool] | None) -> bool: - return ListOperations.all(self, *p) + def all(self, p: ElementPredicate[T]) -> bool: + return ListOperations.all(self, p) def all_is_instance(self, *t: type) -> bool: return ListOperations.all_is_instance(self, *t) - def any(self, *p: Callable[[T], bool] | None) -> bool: - return ListOperations.any(self, *p) + def any(self, p: ElementPredicate[T]) -> bool: + return ListOperations.any(self, p) - def remove_all_if(self, *p: Callable[[]]) -> "CruList"[T]: - return CruList(ListOperations.remove_all_if(self, *p)) + def find_all_if(self, p: ElementPredicate[T]) -> "CruList"[T]: + return CruList(ListOperations.find_all_if(self, p)) - def remove_all_value(self, *r) -> "CruList"[T]: + def find_if(self, p: ElementPredicate[T]) -> T | CRU_NOT_FOUND: + return ListOperations.find_if(self, p) + + def find_all_indices_if(self, p: ElementPredicate[T]) -> "CruList"[int]: + return CruList(ListOperations.find_all_indices_if(self, p)) + + def find_index_if(self, p: ElementPredicate[T]) -> int | CRU_NOT_FOUND: + return ListOperations.find_index_if(self, p) + + def transform(self, *f: OptionalElementTransformer) -> "CruList"[Any]: + return CruList(ListOperations.transform(self, *f)) + + def transform_if(self, f: OptionalElementTransformer, p: ElementPredicate[T]) -> "CruList"[Any]: + return CruList(ListOperations.transform_if(self, f, p)) + + def remove_by_indices(self, *index: int) -> "CruList"[T]: + return CruList(ListOperations.remove_by_indices(self, *index)) + + def remove_if(self, p: ElementPredicate[T]) -> "CruList"[T]: + return CruList(ListOperations.remove_if(self, p)) + + def remove_all_if(self, p: ElementPredicate[T]) -> "CruList"[T]: + return CruList(ListOperations.remove_all_if(self, p)) + + def remove_all_value(self, *r: Any) -> "CruList"[T]: return CruList(ListOperations.remove_all_value(self, *r)) def replace_all_value(self, old_value: Any, new_value: R) -> "CruList"[T | R]: @@ -109,3 +167,123 @@ class CruList(list, Generic[T]): @staticmethod def make(l: CanBeList[T]) -> "CruList"[T]: return CruList(ListOperations.make(l)) + + +class CruInplaceList(CruList, Generic[T]): + + def clear(self) -> "CruInplaceList[T]": + self.clear() + return self + + def extend(self, *l: Iterable[T]) -> "CruInplaceList[T]": + self.extend(l) + return self + + def reset(self, *l: Iterable[T]) -> "CruInplaceList[T]": + self.clear() + self.extend(l) + return self + + def transform(self, *f: OptionalElementTransformer) -> "CruInplaceList"[Any]: + return self.reset(super().transform(*f)) + + def transform_if(self, f: OptionalElementTransformer, p: ElementPredicate[T]) -> "CruInplaceList"[Any]: + return self.reset(super().transform_if(f, p)) + + def remove_by_indices(self, *index: int) -> "CruInplaceList"[T]: + return self.reset(super().remove_by_indices(*index)) + + def remove_all_if(self, p: ElementPredicate[T]) -> "CruInplaceList"[T]: + return self.reset(super().remove_all_if(p)) + + def remove_all_value(self, *r: Any) -> "CruInplaceList"[T]: + return self.reset(super().remove_all_value(*r)) + + def replace_all_value(self, old_value: Any, new_value: R) -> "CruInplaceList"[T | R]: + return self.reset(super().replace_all_value(old_value, new_value)) + + @staticmethod + def make(l: CanBeList[T]) -> "CruInplaceList"[T]: + return CruInplaceList(ListOperations.make(l)) + + +K = TypeVar("K") + + +class CruUniqueKeyInplaceList(Generic[T, K]): + KeyGetter = Callable[[T], K] + + def __init__(self, get_key: KeyGetter, *, before_add: Callable[[T], T] | None = None): + super().__init__() + self._get_key = get_key + self._before_add = before_add + self._l: CruInplaceList[T] = CruInplaceList() + + @property + def object_key_getter(self) -> KeyGetter: + return self._get_key + + @property + def internal_list(self) -> CruInplaceList[T]: + return self._l + + def validate_self(self): + keys = self._l.transform(self._get_key) + if len(keys) != len(set(keys)): + raise ValueError("Duplicate keys!") + + def get_or(self, k: K, fallback: Any = CRU_NOT_FOUND) -> T | Any: + r = self._l.find_if(lambda i: k == self._get_key(i)) + return r if r is not CRU_NOT_FOUND else fallback + + def get(self, k: K) -> T: + v = self.get_or(k, CRU_NOT_FOUND) + if v is CRU_NOT_FOUND: + raise KeyError(f"Key not found!") + return v + + def has_key(self, k: K) -> bool: + return self.get_or(k, CRU_NOT_FOUND) is not CRU_NOT_FOUND + + def has_any_key(self, *k: K) -> bool: + return self._l.any(lambda i: self._get_key(i) in k) + + def try_remove(self, k: K) -> bool: + i = self._l.find_index_if(lambda v: k == self._get_key(v)) + if i is CRU_NOT_FOUND: return False + self._l.remove_by_indices(i) + return True + + def remove(self, k: K, allow_absense: bool = False) -> None: + if not self.try_remove(k) and not allow_absense: + raise KeyError(f"Key {k} not found!") + + def add(self, v: T, /, replace: bool = False) -> None: + if self.has_key(self._get_key(v)): + if replace: + self.remove(self._get_key(v)) + else: + raise ValueError(f"Key {self._get_key(v)} already exists!") + if self._before_add is not None: + v = self._before_add(v) + self._l.append(v) + + def set(self, v: T) -> None: + self.add(v, True) + + def extend(self, l: Iterable[T], /, replace: bool = False) -> None: + if not replace and self.has_any_key([self._get_key(i) for i in l]): + raise ValueError("Keys already exists!") + if self._before_add is not None: + l = [self._before_add(i) for i in l] + keys = [self._get_key(i) for i in l] + self._l.remove_all_if(lambda i: self._get_key(i) in keys).extend(l) + + def clear(self) -> None: + self._l.clear() + + def __iter__(self): + return iter(self._l) + + def __len__(self): + return len(self._l) diff --git a/tools/cru-py/cru/value.py b/tools/cru-py/cru/value.py index cddbde9..7b716f0 100644 --- a/tools/cru-py/cru/value.py +++ b/tools/cru-py/cru/value.py @@ -6,7 +6,7 @@ from abc import abstractmethod, ABCMeta from collections.abc import Mapping, Callable from typing import Any, ClassVar, Literal, TypeVar, Generic, ParamSpec -from .excp import CruInternalLogicError, CruException, CRU_EXCEPTION_ATTR_DEF_REGISTRY +from .excp import CruInternalLogicError, CruException def _str_case_in(s: str, case: bool, l: list[str]) -> bool: @@ -24,10 +24,10 @@ T = TypeVar("T") class _ValueErrorMixin: VALUE_TYPE_KEY = "value_type" - CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_with( + CruException.ATTR_REGISTRY.make_builder( VALUE_TYPE_KEY, "The type of the value that causes the exception." - ) + ).build() class ValidationError(CruException, _ValueErrorMixin): @@ -38,7 +38,7 @@ class ValidationError(CruException, _ValueErrorMixin): @property def value_type(self) -> _ValueTypeForward[T] | None: - return self[ValidationError.VALUE_TYPE_KEY] + return self.get_attr(ValidationError.VALUE_TYPE_KEY) class ValueStringConvertionError(CruException, _ValueErrorMixin): @@ -50,7 +50,7 @@ class ValueStringConvertionError(CruException, _ValueErrorMixin): @property def value_type(self) -> _ValueTypeForward[T] | None: - return self[ValueStringConvertionError.VALUE_TYPE_KEY] + return self.get_attr(ValidationError.VALUE_TYPE_KEY) class ValueType(Generic[T], metaclass=ABCMeta): |