aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--tools/cru-py/cru/attr.py287
-rw-r--r--tools/cru-py/cru/excp.py93
-rw-r--r--tools/cru-py/cru/paths.py2
-rw-r--r--tools/cru-py/cru/service/nginx.py9
-rw-r--r--tools/cru-py/cru/system.py2
-rw-r--r--tools/cru-py/cru/util/__init__.py16
-rw-r--r--tools/cru-py/cru/util/_const.py51
-rw-r--r--tools/cru-py/cru/util/_func.py46
-rw-r--r--tools/cru-py/cru/util/_list.py272
-rw-r--r--tools/cru-py/cru/value.py10
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):
generated by cgit v1.2.3 (git 2.46.0) at 2025-10-31 22:01:50 +0000