aboutsummaryrefslogtreecommitdiff
path: root/tools/cru-py/cru
diff options
context:
space:
mode:
authorcrupest <crupest@outlook.com>2024-11-11 01:12:29 +0800
committerYuqian Yang <crupest@crupest.life>2024-12-18 18:31:27 +0800
commitf35afa17a47fa8e0cafbff5956feab64adb503be (patch)
tree535cc9cfbc23d352f907b485ff23197af5deb5ff /tools/cru-py/cru
parent723c9a963a96b25a7498f3e0417307e89c8bb684 (diff)
downloadcrupest-f35afa17a47fa8e0cafbff5956feab64adb503be.tar.gz
crupest-f35afa17a47fa8e0cafbff5956feab64adb503be.tar.bz2
crupest-f35afa17a47fa8e0cafbff5956feab64adb503be.zip
HALF WORK: 2024.11.10
Diffstat (limited to 'tools/cru-py/cru')
-rw-r--r--tools/cru-py/cru/attr.py359
-rw-r--r--tools/cru-py/cru/excp.py76
-rw-r--r--tools/cru-py/cru/util/__init__.py14
-rw-r--r--tools/cru-py/cru/util/_func.py150
-rw-r--r--tools/cru-py/cru/util/_list.py111
-rw-r--r--tools/cru-py/cru/util/_type.py42
6 files changed, 633 insertions, 119 deletions
diff --git a/tools/cru-py/cru/attr.py b/tools/cru-py/cru/attr.py
index d07cc55..5d4f276 100644
--- a/tools/cru-py/cru/attr.py
+++ b/tools/cru-py/cru/attr.py
@@ -1,8 +1,9 @@
+import copy
from collections.abc import Callable
-from dataclasses import dataclass
-from types import NoneType
+from dataclasses import dataclass, field
from typing import Any
-from copy import deepcopy
+
+from .util import CanBeList, TypeSet, L, ListOperations
@dataclass
@@ -12,63 +13,243 @@ class CruAttr:
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__()
+
+
+def _is_attr_void(v: Any) -> bool:
+ return isinstance(v, _CruAttrVoid)
+
+
@dataclass
class CruAttrDef:
name: str
- default_description: str
- allow_types: None | set[type]
- allow_none: bool
- default_value: Any
- transformer: Callable[[Any], Any] | None
+ description: str
+ default_factory: Callable[[], Any]
+ transformer: Callable[[Any], Any]
validator: Callable[[Any], None]
- def __init__(self, name: str, default_description: str, *,
- allow_types: list[type] | type | None, allow_none: bool, default_value: Any = None,
- transformer: Callable[[Any], Any] | None = None,
- validator: Callable[[Any], None] | None = None) -> None:
+ def __init__(self, name: str, description: str, default_factory: Callable[[], Any],
+ transformer: Callable[[Any], Any], validator: Callable[[Any], None]) -> None:
self.name = name
- self.default_description = default_description
- self.default_value = default_value
- #TODO: CONTINUE TOMORROW
- if allow_types is None:
- allow_types = []
- elif isinstance(allow_types, type):
- allow_types = [allow_types]
- else:
- for t in allow_types:
- if not isinstance(t, type):
- raise TypeError(f"Invalid value of python type : {t}")
- self.allow_types = set(filter(lambda tt: tt is not NoneType, allow_types))
- self.allow_none = allow_none
+ self.description = description
+ self.default_factory = default_factory
self.transformer = transformer
self.validator = validator
- self.default_value = self.transform_and_validate(self.default_value)
- self.default_value = deepcopy(self.default_value)
def transform(self, value: Any) -> Any:
if self.transformer is not None:
return self.transformer(value)
return value
- def validate(self, value: Any, /, override_allow_none: bool | None = None) -> None:
- allow_none = override_allow_none if override_allow_none is not None else self.allow_none
- if value is None and not allow_none:
- raise TypeError(f"None is not allowed!")
- if len(self.allow_types) != 0 and type(value) not in self.allow_types:
- raise TypeError(f"Type of {value} is not allowed!")
+ 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:
- return self.validator(value)
- return None
+ self.validator(value)
- def transform_and_validate(self, value: Any, /, override_allow_none: bool | None = None) -> Any:
+ def transform_and_validate(self, value: Any = CRU_ATTR_USE_DEFAULT, /, force_allow_none: bool = False) -> Any:
value = self.transform(value)
- self.validate(value, override_allow_none)
+ self.validate(value, force_allow_none)
return value
- def make(self, value: Any, description: None | str = None) -> CruAttr:
+ def make_default_value(self) -> Any:
+ return self.transform_and_validate(self.default_factory())
+
+ def adopt(self, attr: CruAttr) -> CruAttr:
+ attr = copy.deepcopy(attr)
+
+ if attr.name is None:
+ attr.name = self.name
+ 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:
+ attr.value = self.make_default_value()
+ else:
+ attr.value = self.transform_and_validate(attr.value)
+
+ if attr.description is None:
+ attr.description = self.description
+
+ 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
value = self.transform_and_validate(value)
- return CruAttr(self.name, value if value is not None else deepcopy(self.default_value),
- description if description is not None else self.default_description)
+ return CruAttr(self.name, value,
+ description if description is not None else self.description)
+
+
+@dataclass
+class CruAttrDefBuilder:
+ 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)
+ 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)
+
+ build_hook: Callable[[CruAttrDef], None] | None = field(default=None)
+
+ def __init__(self, name: str, description: str) -> None:
+ super().__init__()
+ self.name = name
+ self.description = description
+
+ def auto_adjust_default(self) -> "CruAttrDefBuilder":
+ if self.default is not CRU_ATTR_NOT_SET and self.default is not None:
+ return self
+ if self.allow_none and _is_attr_void(self.default):
+ self.default = None
+ if not self.allow_none and self.default is None:
+ self.default = CRU_ATTR_NOT_SET
+ if self.auto_list and not self.allow_none:
+ self.default = []
+
+ def with_name(self, name: str) -> "CruAttrDefBuilder":
+ self.name = name
+ return self
+
+ def with_description(self, default_description: str) -> "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
+ return self
+
+ def with_default_factory(self, default_factory: Callable[[], Any]) -> "CruAttrDefBuilder":
+ return self.with_default(default_factory, is_factory=True)
+
+ def with_types(self, allowed_types: CanBeList[type], default: Any = CRU_ATTR_NOT_SET) -> "CruAttrDefBuilder":
+ self.types = allowed_types
+ if _is_attr_void(default):
+ self.default = default
+ return self
+
+ def with_allow_none(self, allow_none: bool, default: Any = CRU_ATTR_NOT_SET) -> "CruAttrDefBuilder":
+ self.allow_none = allow_none
+ if _is_attr_void(default):
+ self.default = default
+ return self
+
+ def with_optional(self, default: Any = CRU_ATTR_NOT_SET) -> "CruAttrDefBuilder":
+ return self.with_allow_none(True, default)
+
+ def with_required(self, default: Any = CRU_ATTR_NOT_SET) -> "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_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":
+ if transformer is not None:
+ self.transformers.append(transformer)
+ return self
+
+ def clear_transformers(self) -> "CruAttrDefBuilder":
+ self.transformers.clear()
+ return self
+
+ def add_validator(self, validator: Callable[[Any], None] | None) -> "CruAttrDefBuilder":
+ if validator is not None:
+ self.validators.append(validator)
+ return self
+
+ def clear_validators(self) -> "CruAttrDefBuilder":
+ self.validators.clear()
+ return self
+
+ def with_override_transformer(self, override_transformer: Callable[[Any], Any] | None) -> "CruAttrDefBuilder":
+ self.override_transformer = override_transformer
+ return self
+
+ def with_override_validator(self, override_validator: Callable[[Any], None] | None) -> "CruAttrDefBuilder":
+ self.override_validator = override_validator
+ return self
+
+ def is_valid(self) -> tuple[bool, str]:
+ if not isinstance(self.name, str):
+ return False, "Name must be a string!"
+ if not isinstance(self.description, str):
+ return False, "Default description must be a string!"
+ if not self.allow_none and self.default is None and self.default_factory is None:
+ return False, "Default must be set if allow_none is False!"
+
+ @staticmethod
+ def _build(b: "CruAttrDefBuilder", auto_adjust_default: bool = True) -> CruAttrDef:
+ if auto_adjust_default:
+ b.auto_adjust_default()
+
+ valid, err = b.is_valid()
+ if not valid: raise ValueError(err)
+
+ def composed_transformer(v: Any):
+ if b.auto_list:
+ v = L.make(v)
+ for t in b.transformers:
+ v = t(v)
+ return v
+
+ type_set = TypeSet(b.types)
+
+ def composed_validator(v: Any):
+ if b.auto_list:
+ type_set.check_value_list(v, allow_none=b.allow_none)
+ ListOperations.foreach(v, *b.validators)
+ else:
+ type_set.check_value(v, allow_none=b.allow_none)
+ for vl in b.validators: vl(v)
+
+ validator = b.override_validator or composed_validator
+ transformer = b.override_transformer or composed_transformer
+
+ default_factory = b.default_factory
+ if b.default_factory is None:
+ default_factory = lambda: copy.deepcopy(b.default)
+ validator(default_factory())
+
+ d = CruAttrDef(b.name, b.description, default_factory, transformer, validator)
+ if b.build_hook: b.build_hook(d)
+ return d
+
+ def build(self, auto_adjust_default=True) -> CruAttrDef:
+ c = copy.deepcopy(self)
+ self.build_hook = None
+ return CruAttrDefBuilder._build(c, auto_adjust_default)
class CruAttrDefRegistry:
@@ -86,34 +267,10 @@ class CruAttrDefRegistry:
raise ValueError(f"Attribute {def_.name} already exists!")
self._def_list.append(def_)
- def register_with(self, name: str, default_description: str, *,
- allow_types: list[type] | type | None, allow_none: bool,
- default_value: Any = None,
- transformer: Callable[[Any], Any] | None = None,
- validator: Callable[[Any], None] | None = None
- ) -> CruAttrDef:
- def_ = CruAttrDef(name, default_description, default_value=default_value, allow_types=allow_types,
- allow_none=allow_none, transformer=transformer, validator=validator)
- self.register(def_)
- return def_
-
- def register_required(self, name: str, default_description: str, *,
- allow_types: list[type] | type | None,
- default_value: Any = None,
- transformer: Callable[[Any], Any] | None = None,
- validator: Callable[[Any], None] | None = None
- ) -> CruAttrDef:
- return self.register_with(name, default_description, default_value=default_value, allow_types=allow_types,
- allow_none=False, transformer=transformer, validator=validator)
-
- def register_optional(self, name: str, default_description: str, *,
- allow_types: list[type] | type | None,
- default_value: Any = None,
- transformer: Callable[[Any], Any] | None = None,
- validator: Callable[[Any], None] | None = None
- ) -> CruAttrDef:
- return self.register_with(name, default_description, default_value=default_value, allow_types=allow_types,
- allow_none=True, transformer=transformer, validator=validator)
+ def make_builder(self, name: str, default_description: str) -> CruAttrDefBuilder:
+ b = CruAttrDefBuilder(name, default_description)
+ b.build_hook = lambda a: self.register(a)
+ return b
def get_item_optional(self, name: str) -> CruAttrDef | None:
for i in self._def_list:
@@ -121,5 +278,67 @@ class CruAttrDefRegistry:
return i
return None
- def __getitem__(self, item) -> CruAttrDef | None:
- return self.get_item_optional(item)
+
+class CruAttrTable:
+ 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:
+ return fallback
+ return a.value
+
+ def get_value(self, name: str) -> Any:
+ a = self.get_attr(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 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 _check_already_exist(self, name: str) -> None:
+ if name in self:
+ raise KeyError(f"Attribute {name} already exists!")
+
+ 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 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
+
+ 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
+
+ def __contains__(self, name):
+ return any(a.name == name for a in self._attrs)
diff --git a/tools/cru-py/cru/excp.py b/tools/cru-py/cru/excp.py
index 5a5871b..6734140 100644
--- a/tools/cru-py/cru/excp.py
+++ b/tools/cru-py/cru/excp.py
@@ -1,63 +1,41 @@
-from collections.abc import Callable
-from dataclasses import dataclass
-from types import NoneType
from typing import Any
-from cru.attr import CruAttrDefRegistry
+from .attr import CruAttrDefRegistry, CRU_ATTR_USE_DEFAULT, CruAttr, CruAttrTable
+from .util import CanBeList
CRU_EXCEPTION_ATTR_DEF_REGISTRY = CruAttrDefRegistry()
-
-
-class CruException(Exception):
- @staticmethod
- def transform_inner(inner: Exception | list[Exception] | None):
- if inner is None:
- return None
- if isinstance(inner, Exception):
- return [inner]
- if isinstance(inner, list):
- return inner
-
- @staticmethod
- def validate_inner(inner: list[Exception]):
- for i in inner:
- if not isinstance(i, Exception):
- raise TypeError(f"Invalid inner exception: {i}")
-
- MESSAGE_DEF = CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_required("message", "Message describing the exception.",
- allow_types=str, default_value="")
- INNER_DEF = CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_required("inner", "Inner exception.",
- allow_types=list, default_value=[],
- transformer=transform_inner, validator=validate_inner)
- INTERNAL_DEF = CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_required("internal",
- "True if the exception is caused by wrong internal logic. False if it is caused by user's wrong input.",
- allow_types=bool, default_value=False)
- CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_optional("name", "Name of the object that causes the exception.",
- allow_types=str)
- CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_optional("value", "Value that causes the exception.",
- allow_types=[])
- CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_with("path", "Path that causes the exception.",)
- CRU_EXCEPTION_ATTR_DEF_REGISTRY.register_with("type", "Python type related to the exception.")
-
+_REGISTRY = CRU_EXCEPTION_ATTR_DEF_REGISTRY
+
+
+class CruException(Exception, CruAttrTable):
+ MESSAGE_KEY = "message"
+ INNER_KEY = "inner"
+ INTERNAL_KEY = "internal"
+ NAME_KEY = "name"
+ VALUE_KEY = "value"
+ 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(
+ 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()
+
+ # TODO: CONTINUE HERE TOMORROW!
def __init__(self, message: str, *args,
- inner: Exception | list[Exception] | None = None,
- internal: bool = False,
- name: str | None = None,
- value: Any | None = None,
- path: str | None = None,
- type_: type | None = None,
init_attrs: dict[str, Any] | None = None,
attrs: dict[str, Any] | None = None, **kwargs) -> None:
super().__init__(message, *args)
self._attrs = {
CruException.MESSAGE_KEY: message,
- CruException.INTERNAL_KEY: internal,
- CruException.INNER_KEY: inner,
- CruException.NAME_KEY: name,
- CruException.VALUE_KEY: value,
- CruException.PATH_KEY: path,
- CruException.TYPE_KEY: type_,
}
if init_attrs is not None:
self._attrs.update(init_attrs)
diff --git a/tools/cru-py/cru/util/__init__.py b/tools/cru-py/cru/util/__init__.py
new file mode 100644
index 0000000..83ab9f3
--- /dev/null
+++ b/tools/cru-py/cru/util/__init__.py
@@ -0,0 +1,14 @@
+from ._func import CruFunction, MetaFunction, RawFunctions, WrappedFunctions, PLACEHOLDER
+from ._list import CruList, ListOperations, CanBeList
+from ._type import TypeSet
+
+F = CruFunction
+WF = WrappedFunctions
+L = CruList
+
+__all__ = [
+ "CruFunction", "MetaFunction", "RawFunctions", "WrappedFunctions", "PLACEHOLDER",
+ "CruList", "ListOperations", "CanBeList",
+ "TypeSet",
+ "F", "WF", "L",
+]
diff --git a/tools/cru-py/cru/util/_func.py b/tools/cru-py/cru/util/_func.py
new file mode 100644
index 0000000..5fb49a9
--- /dev/null
+++ b/tools/cru-py/cru/util/_func.py
@@ -0,0 +1,150 @@
+from collections.abc import Callable, Iterable
+from typing import TypeVar, Any, ParamSpec
+
+from ._list import ListOperations, CruList
+
+T = TypeVar("T")
+R = TypeVar("R")
+R1 = TypeVar("R1")
+P = ParamSpec("P")
+P1 = ParamSpec("P1")
+
+
+class _Placeholder:
+ pass
+
+
+PLACEHOLDER = _Placeholder()
+
+
+class RawFunctions:
+ @staticmethod
+ def ignore(*_v, **_kwargs) -> None:
+ return None
+
+ @staticmethod
+ def true_(*_v, **_kwargs) -> True:
+ return True
+
+ @staticmethod
+ 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 equal(a: Any, b: Any) -> bool:
+ return a == b
+
+ @staticmethod
+ def not_equal(a: Any, b: Any) -> bool:
+ return a != b
+
+ @staticmethod
+ def not_(v):
+ return not v
+
+
+class MetaFunction:
+ @staticmethod
+ def bind(f: Callable[P, R], *bind_args, **bind_kwargs) -> Callable[P1, R1]:
+ def bound(*args, **kwargs):
+ popped = 0
+ real_args = []
+ for a in bind_args:
+ if isinstance(a, _Placeholder):
+ real_args.append(args[popped])
+ popped += 1
+ else:
+ real_args.append(a)
+ real_args.extend(args[popped:])
+ return f(*real_args, **(bind_kwargs | kwargs))
+
+ return bound
+
+ @staticmethod
+ def chain(*fs: Callable) -> Callable:
+ if len(fs) == 0:
+ raise ValueError("At least one function is required!")
+ rf = fs[0]
+ for f in fs[1:]:
+ def n(*args, **kwargs):
+ r = rf(*args, **kwargs)
+ r = r if isinstance(r, tuple) else (r,)
+ return f(*r)
+
+ rf = n
+ return rf
+
+ @staticmethod
+ def chain_single(f: Callable[P, R], f1: Callable[P1, R1], *bind_args, **bind_kwargs) -> \
+ Callable[
+ P, R1]:
+ return MetaFunction.chain(f, MetaFunction.bind(f1, *bind_args, **bind_kwargs))
+
+ convert_r = chain_single
+
+ @staticmethod
+ def neg(f: Callable[P, bool]) -> Callable[P, bool]:
+ return MetaFunction.convert_r(f, RawFunctions.not_)
+
+
+# Advanced Function Wrapper
+class CruFunction:
+ def __init__(self, f):
+ self._f = f
+
+ @property
+ def f(self) -> Callable:
+ return self._f
+
+ def bind(self, *bind_args, **bind_kwargs) -> "CruFunction":
+ self._f = MetaFunction.bind(self._f, *bind_args, **bind_kwargs)
+ return self
+
+ def chain(self, *fs: Callable) -> "CruFunction":
+ self._f = MetaFunction.chain(self._f, *fs)
+ return self
+
+ def chain_single(self, f: Callable[P, R], f1: Callable[P1, R1], *bind_args,
+ **bind_kwargs) -> "CruFunction":
+ self._f = MetaFunction.chain_single(self._f, f, f1, *bind_args, **bind_kwargs)
+ return self
+
+ def convert_r(self, f: Callable[P, R], f1: Callable[P1, R1], *bind_args,
+ **bind_kwargs) -> "CruFunction":
+ self._f = MetaFunction.convert_r(self._f, f, f1, *bind_args, **bind_kwargs)
+ return self
+
+ def neg(self) -> "CruFunction":
+ self._f = MetaFunction.neg(self._f)
+ return self
+
+ 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))
+
+
+class WrappedFunctions:
+ identity = CruFunction(RawFunctions.identity)
+ ignore = CruFunction(RawFunctions.ignore)
+ 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
new file mode 100644
index 0000000..3b4cb28
--- /dev/null
+++ b/tools/cru-py/cru/util/_list.py
@@ -0,0 +1,111 @@
+from collections.abc import Iterable, Callable
+from typing import TypeVar, ParamSpec, Any, Generic
+
+T = TypeVar("T")
+R = TypeVar("R")
+P = ParamSpec("P")
+
+CanBeList = T | Iterable[T] | None
+
+
+class ListOperations:
+
+ @staticmethod
+ def foreach(l: Iterable[T], *f: Callable[[T], None] | None) -> None:
+ if len(f) == 0: return
+ for v in l:
+ for f_ in f:
+ if f_ is not None:
+ f_(v)
+
+ @staticmethod
+ def transform(l: Iterable[T], *f: Callable | None) -> list[R]:
+ r = []
+ for v in l:
+ for f_ in f:
+ if f_ is not None:
+ v = f_(v)
+ r.append(v)
+ return r
+
+ @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)]
+
+ @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!")
+ for v in l:
+ for p_ in p:
+ if p_ is not None and not p_(v): return False
+ return True
+
+ @staticmethod
+ def all_is_instance(l: Iterable[T], *t: type) -> bool:
+ return all(type(v) in t for v in l)
+
+ @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!")
+ for v in l:
+ for p_ in p:
+ if p_ is not None and p_(v): return True
+ return False
+
+ @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]
+
+ @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
+
+ return [v for v in l if stay(v)]
+
+ @staticmethod
+ def remove_all_value(l: Iterable[T], *r) -> 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]
+
+
+class CruList(list, Generic[T]):
+
+ def foreach(self, *f: Callable[[T], None] | None) -> None:
+ ListOperations.foreach(self, *f)
+
+ def transform(self, *f: Callable[[T], R] | None) -> "CruList"[R]:
+ return CruList(ListOperations.transform(self, *f))
+
+ 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 all(self, *p: Callable[[T], bool] | None) -> 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 remove_all_if(self, *p: Callable[[]]) -> "CruList"[T]:
+ return CruList(ListOperations.remove_all_if(self, *p))
+
+ def remove_all_value(self, *r) -> "CruList"[T]:
+ return CruList(ListOperations.remove_all_value(self, *r))
+
+ def replace_all_value(self, old_value: Any, new_value: R) -> "CruList"[T | R]:
+ return CruList(ListOperations.replace_all_value(self, old_value, new_value))
+
+ @staticmethod
+ def make(l: CanBeList[T]) -> "CruList"[T]:
+ return CruList(ListOperations.make(l))
diff --git a/tools/cru-py/cru/util/_type.py b/tools/cru-py/cru/util/_type.py
new file mode 100644
index 0000000..dc50def
--- /dev/null
+++ b/tools/cru-py/cru/util/_type.py
@@ -0,0 +1,42 @@
+from types import NoneType
+from typing import Any
+
+from ._list import CanBeList, CruList
+
+DEFAULT_NONE_ERR = ValueError
+DEFAULT_NONE_ERR_MSG = "None is not allowed here."
+DEFAULT_TYPE_ERR = ValueError
+DEFAULT_TYPE_ERR_MSG = "Type of object is not allowed here."
+
+
+class TypeSet(set[type]):
+ def __init__(self, *l: type):
+ l = CruList.make(l).remove_all_value(None, NoneType)
+ if not l.all_is_instance(type):
+ raise TypeError("t must be a type or None.")
+ super().__init__(l)
+
+ def check_value(self, v: Any, /, allow_none: bool, empty_allow_all: bool = True, *,
+ none_err: type[Exception] = DEFAULT_NONE_ERR,
+ none_err_msg: str = DEFAULT_NONE_ERR_MSG,
+ type_err: type[Exception] = DEFAULT_TYPE_ERR,
+ type_err_msg: str = DEFAULT_TYPE_ERR_MSG) -> None:
+ if v is None:
+ if allow_none:
+ return
+ else:
+ raise none_err(none_err_msg)
+ if len(self) == 0 and empty_allow_all:
+ return
+ if type(v) not in self:
+ raise type_err(type_err_msg)
+
+ def check_value_list(self, l: CanBeList, /, allow_none: bool, empty_allow_all: bool = True, *,
+ none_err: type[Exception] = DEFAULT_NONE_ERR,
+ none_err_msg: str = DEFAULT_NONE_ERR_MSG,
+ type_err: type[Exception] = DEFAULT_TYPE_ERR,
+ type_err_msg: str = DEFAULT_TYPE_ERR_MSG) -> None:
+ l = CruList.make(l)
+ for v in l:
+ self.check_value(v, allow_none, empty_allow_all, none_err=none_err, none_err_msg=none_err_msg,
+ type_err=type_err, type_err_msg=type_err_msg)