aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--tools/cru-py/cru/_base.py17
-rw-r--r--tools/cru-py/cru/_event.py50
-rw-r--r--tools/cru-py/cru/_func.py3
-rw-r--r--tools/cru-py/cru/_iter.py180
-rw-r--r--tools/cru-py/cru/_list.py143
-rw-r--r--tools/cru-py/cru/_type.py59
-rw-r--r--tools/cru-py/cru/attr.py335
-rw-r--r--tools/cru-py/cru/config.py2
-rw-r--r--tools/cru-py/cru/error.py27
-rw-r--r--tools/cru-py/cru/excp.py116
-rw-r--r--tools/cru-py/cru/paths.py5
-rw-r--r--tools/cru-py/cru/value.py157
12 files changed, 527 insertions, 567 deletions
diff --git a/tools/cru-py/cru/_base.py b/tools/cru-py/cru/_base.py
index 96466d1..2310bfb 100644
--- a/tools/cru-py/cru/_base.py
+++ b/tools/cru-py/cru/_base.py
@@ -1,24 +1,13 @@
-from typing import Any, NoReturn
+from typing import Any
from ._lang import remove_none
+from .error import CruInternalError
-class CruException(Exception):
- """Base exception class of all exceptions in cru."""
-
-
-class CruNamespaceError(CruException):
+class CruNamespaceError(CruInternalError):
"""Raised when a namespace is not found."""
-class CruUnreachableError(CruException):
- """Raised when a code path is unreachable."""
-
-
-def cru_unreachable() -> NoReturn:
- raise CruUnreachableError()
-
-
class _Cru:
NAME_PREFIXES = ("CRU_", "Cru", "cru_")
diff --git a/tools/cru-py/cru/_event.py b/tools/cru-py/cru/_event.py
index ee914f0..65265fd 100644
--- a/tools/cru-py/cru/_event.py
+++ b/tools/cru-py/cru/_event.py
@@ -1,24 +1,28 @@
-from typing import ParamSpec, TypeVar, Callable
+from __future__ import annotations
-from ._iter import CruInplaceList, CruList
+from collections.abc import Callable
+from typing import Generic, ParamSpec, TypeVar
-P = ParamSpec('P')
-R = TypeVar('R')
-F = Callable[P, R]
+from ._list import CruList
+_P = ParamSpec("_P")
+_R = TypeVar("_R")
-class EventHandlerToken:
- def __init__(self, event: "Event", handler: F, once: bool = False) -> None:
+
+class EventHandlerToken(Generic[_P, _R]):
+ def __init__(
+ self, event: Event, handler: Callable[_P, _R], once: bool = False
+ ) -> None:
self._event = event
self._handler = handler
self._once = once
@property
- def event(self) -> "Event":
+ def event(self) -> Event:
return self._event
@property
- def handler(self) -> F:
+ def handler(self) -> Callable[_P, _R]:
return self._handler
@property
@@ -26,16 +30,32 @@ class EventHandlerToken:
return self._once
-class Event:
+class Event(Generic[_P, _R]):
def __init__(self, name: str) -> None:
self._name = name
- self._tokens: CruInplaceList[EventHandlerToken] = CruInplaceList()
+ self._tokens: CruList[EventHandlerToken] = CruList()
- def register(self, handler: F, once: bool = False) -> EventHandlerToken:
+ def register(
+ self, handler: Callable[_P, _R], once: bool = False
+ ) -> EventHandlerToken:
token = EventHandlerToken(self, handler, once)
self._tokens.append(token)
return token
- def unregister(self, *h: EventHandlerToken | F) -> int:
-
- self._tokens.find_all_indices_if(lambda t: )
+ def unregister(self, *handlers: EventHandlerToken | Callable[_P, _R]) -> int:
+ old_length = len(self._tokens)
+ self._tokens.reset(
+ self._tokens.as_cru_iterator().filter(
+ (lambda t: t in handlers or t.handler in handlers)
+ )
+ )
+ return old_length - len(self._tokens)
+
+ def trigger(self, *args: _P.args, **kwargs: _P.kwargs) -> CruList[_R]:
+ results = CruList(
+ self._tokens.as_cru_iterator()
+ .transform(lambda t: t.handler(*args, **kwargs))
+ .to_list()
+ )
+ self._tokens.reset(self._tokens.as_cru_iterator().filter(lambda t: not t.once))
+ return results
diff --git a/tools/cru-py/cru/_func.py b/tools/cru-py/cru/_func.py
index 2b317ab..fc57802 100644
--- a/tools/cru-py/cru/_func.py
+++ b/tools/cru-py/cru/_func.py
@@ -1,11 +1,10 @@
from __future__ import annotations
-from collections.abc import Callable
+from collections.abc import Callable, Iterable
from enum import Flag, auto
from typing import (
Any,
Generic,
- Iterable,
Literal,
ParamSpec,
TypeAlias,
diff --git a/tools/cru-py/cru/_iter.py b/tools/cru-py/cru/_iter.py
index bf1dfde..b91195b 100644
--- a/tools/cru-py/cru/_iter.py
+++ b/tools/cru-py/cru/_iter.py
@@ -1,12 +1,10 @@
from __future__ import annotations
-from collections.abc import Iterable, Callable
+from collections.abc import Iterable, Callable, Generator, Iterator
from dataclasses import dataclass
from enum import Enum
from typing import (
Concatenate,
- Generator,
- Iterator,
Literal,
Never,
Self,
@@ -114,8 +112,8 @@ class _Generic:
o: _O,
max_depth: int = -1,
/,
- is_leave: _Wrapper.ElementPredicate[_O] = _is_not_iterable,
- get_children: _Wrapper.ElementTransformer[_O, Iterable[_O]] = _return_self,
+ is_leave: CruIterator.ElementPredicate[_O] = _is_not_iterable,
+ get_children: CruIterator.ElementTransformer[_O, Iterable[_O]] = _return_self,
*,
_depth: int = 0,
) -> Iterable[_O]:
@@ -241,9 +239,9 @@ class _Generic:
cru_unreachable()
-class _Helper:
+class _Helpers:
@staticmethod
- def with_count(c: Callable[Concatenate[int, _P], _O]) -> Callable[_P, _O]:
+ def auto_count(c: Callable[Concatenate[int, _P], _O]) -> Callable[_P, _O]:
count = 0
def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _O:
@@ -257,49 +255,42 @@ class _Helper:
class _Creators:
- @staticmethod
- def empty(
- *,
- wrapper_type: type["_Wrapper[Never]"] | None = None,
- attr: dict[str, Any] | None = None,
- ) -> _Wrapper[Never]:
- wrapper_type = wrapper_type or _Wrapper
- return wrapper_type(iter([]), attr)
+ class Raw:
+ @staticmethod
+ def empty() -> Iterator[Never]:
+ return iter([])
- @staticmethod
- def range(
- *args,
- wrapper_type: type["_Wrapper[int]"] | None = None,
- attr: dict[str, Any] | None = None,
- ) -> _Wrapper[int]:
- wrapper_type = wrapper_type or _Wrapper
- return wrapper_type(iter(range(*args)), attr)
+ @staticmethod
+ def range(*args) -> Iterator[int]:
+ return iter(range(*args))
- @staticmethod
- def unite(
- *args: _O,
- wrapper_type: type["_Wrapper[_O]"] | None = None,
- attr: dict[str, Any] | None = None,
- ) -> _Wrapper[_O]:
- wrapper_type = wrapper_type or _Wrapper
- return wrapper_type(iter(args), attr)
+ @staticmethod
+ def unite(*args: _T) -> Iterator[_T]:
+ return iter(args)
- @staticmethod
- def _concat(*iterables: Iterable[_T]) -> Iterable[_T]:
- for iterable in iterables:
- yield from iterable
+ @staticmethod
+ def _concat(*iterables: Iterable[_T]) -> Iterable[_T]:
+ for iterable in iterables:
+ yield from iterable
+
+ @staticmethod
+ def concat(*iterables: Iterable[_T]) -> Iterator[_T]:
+ return iter(_Creators.Raw._concat(*iterables))
@staticmethod
- def concat(
- *iterables: Iterable[_T],
- wrapper_type: type["_Wrapper[_T]"] | None = None,
- attr: dict[str, Any] | None = None,
- ) -> _Wrapper[_T]:
- wrapper_type = wrapper_type or _Wrapper
- return wrapper_type(_Creators._concat(*iterables), attr)
+ def _wrap(f: Callable[_P, Iterable[_O]]) -> Callable[_P, CruIterator[_O]]:
+ def _wrapped(*args: _P.args, **kwargs: _P.kwargs) -> CruIterator[_O]:
+ return CruIterator(f(*args, **kwargs))
+
+ return _wrapped
+ empty = _wrap(Raw.empty)
+ range = _wrap(Raw.range)
+ unite = _wrap(Raw.unite)
+ concat = _wrap(Raw.concat)
-class _Wrapper(Generic[_T]):
+
+class CruIterator(Generic[_T]):
ElementOperation: TypeAlias = Callable[[_V], Any]
ElementPredicate: TypeAlias = Callable[[_V], bool]
AnyElementPredicate: TypeAlias = ElementPredicate[Any]
@@ -307,34 +298,26 @@ class _Wrapper(Generic[_T]):
SelfElementTransformer: TypeAlias = ElementTransformer[_V, _V]
AnyElementTransformer: TypeAlias = ElementTransformer[Any, Any]
- def __init__(
- self, iterable: Iterable[_T], attr: dict[str, Any] | None = None
- ) -> None:
- self._iterable = iterable
- self._attr = attr or {}
-
- def __iter__(self) -> Iterator[_T]:
- return self._iterable.__iter__()
+ Creators: TypeAlias = _Creators
+ Helpers: TypeAlias = _Helpers
- @property
- def me(self) -> Iterable[_T]:
- return self._iterable
+ def __init__(self, iterable: Iterable[_T]) -> None:
+ self._iterator = iter(iterable)
- @property
- def my_attr(self) -> dict[str, Any]:
- return self._attr
+ def __iter__(self) -> Iterator[_T]:
+ return self._iterator
- def create_with_me(self, iterable: Iterable[_O]) -> _Wrapper[_O]:
- return type(self)(iterable, self._attr) # type: ignore
+ def create_new_me(self, iterable: Iterable[_O]) -> CruIterator[_O]:
+ return type(self)(iterable) # type: ignore
@staticmethod
def _wrap(
- f: Callable[Concatenate[_Wrapper[_T], _P], Iterable[_O]]
- ) -> Callable[Concatenate[_Wrapper[_T], _P], _Wrapper[_O]]:
+ f: Callable[Concatenate[CruIterator[_T], _P], Iterable[_O]]
+ ) -> Callable[Concatenate[CruIterator[_T], _P], CruIterator[_O]]:
def _wrapped(
- self: _Wrapper[_T], *args: _P.args, **kwargs: _P.kwargs
- ) -> _Wrapper[_O]:
- return self.create_with_me(f(self, *args, **kwargs))
+ self: CruIterator[_T], *args: _P.args, **kwargs: _P.kwargs
+ ) -> CruIterator[_O]:
+ return self.create_new_me(f(self, *args, **kwargs))
return _wrapped
@@ -342,80 +325,79 @@ class _Wrapper(Generic[_T]):
def replace_me(self, iterable: Iterable[_O]) -> Iterable[_O]:
return iterable
- def replace_me_with_empty(self) -> _Wrapper[Never]:
- return _Creators.empty(wrapper_type=type(self), attr=self._attr) # type: ignore
+ def replace_me_with_empty(self) -> CruIterator[Never]:
+ return self.create_new_me(_Creators.Raw.empty())
- def replace_me_with_range(self, *args) -> _Wrapper[int]:
- return _Creators.range(*args, attr=self._attr)
+ def replace_me_with_range(self, *args) -> CruIterator[int]:
+ return self.create_new_me(_Creators.Raw.range(*args))
- def replace_me_with_unite(self, *args: _O) -> _Wrapper[_O]:
- return _Creators.unite(*args, attr=self._attr)
+ def replace_me_with_unite(self, *args: _O) -> CruIterator[_O]:
+ return self.create_new_me(_Creators.Raw.unite(*args))
- def replace_me_with_concat(self, *iterables: Iterable[_T]) -> _Wrapper[_T]:
- return _Creators.concat(*iterables, attr=self._attr)
+ def replace_me_with_concat(self, *iterables: Iterable[_T]) -> CruIterator[_T]:
+ return self.create_new_me(_Creators.Raw.concat(*iterables))
- def to_set(self, discard: Iterable[Any]) -> set[_T]:
- return set(self.me) - set(discard)
+ def to_set(self) -> set[_T]:
+ return set(self)
- def to_list(self, discard: Iterable[Any]) -> list[_T]:
- return [v for v in self.me if v not in set(discard)]
+ def to_list(self) -> list[_T]:
+ return list(self)
def all(self, predicate: ElementPredicate[_T]) -> bool:
- for value in self.me:
+ for value in self:
if not predicate(value):
return False
return True
def any(self, predicate: ElementPredicate[_T]) -> bool:
- for value in self.me:
+ for value in self:
if predicate(value):
return True
return False
def foreach(self, operation: ElementOperation[_T]) -> None:
- for value in self.me:
+ for value in self:
operation(value)
@_wrap
def transform(self, transformer: ElementTransformer[_T, _O]) -> Iterable[_O]:
- for value in self.me:
+ for value in self:
yield transformer(value)
map = transform
@_wrap
def filter(self, predicate: ElementPredicate[_T]) -> Iterable[_T]:
- for value in self.me:
+ for value in self:
if predicate(value):
yield value
@_wrap
def continue_if(self, predicate: ElementPredicate[_T]) -> Iterable[_T]:
- for value in self.me:
+ for value in self:
yield value
if not predicate(value):
break
- def first_n(self, max_count: int) -> _Wrapper[_T]:
+ def first_n(self, max_count: int) -> CruIterator[_T]:
if max_count < 0:
raise ValueError("max_count must be 0 or positive.")
if max_count == 0:
return self.replace_me_with_empty() # type: ignore
- return self.continue_if(_Helper.with_count(lambda i, _: i < max_count - 1))
+ return self.continue_if(_Helpers.auto_count(lambda i, _: i < max_count - 1))
- def drop_n(self, n: int) -> _Wrapper[_T]:
+ def drop_n(self, n: int) -> CruIterator[_T]:
if n < 0:
raise ValueError("n must be 0 or positive.")
if n == 0:
return self
- return self.filter(_Helper.with_count(lambda i, _: i < n))
+ return self.filter(_Helpers.auto_count(lambda i, _: i < n))
def single_or(
- self, fallback: _O | CruNotFound | None = CruNotFound.VALUE
- ) -> _T | _O | Any | CruNotFound:
+ self, fallback: _O | CruNotFound = CruNotFound.VALUE
+ ) -> _T | _O | CruNotFound:
first_2 = self.first_n(2)
has_value = False
- value = None
for element in first_2:
if has_value:
raise ValueError("More than one value found.")
@@ -426,18 +408,23 @@ class _Wrapper(Generic[_T]):
else:
return fallback
+ def first_or(
+ self, fallback: _O | CruNotFound = CruNotFound.VALUE
+ ) -> _T | _O | CruNotFound:
+ return self.first_n(1).single_or(fallback)
+
@_wrap
def flatten(self) -> Iterable[_T | Iterable[_T]]:
- return _Generic.iterable_flatten(self.me)
+ return _Generic.iterable_flatten(self)
- def select_by_indices(self, indices: Iterable[int]) -> _Wrapper[_T]:
+ def select_by_indices(self, indices: Iterable[int]) -> CruIterator[_T]:
index_set = set(indices)
max_index = max(index_set)
return self.first_n(max_index + 1).filter(
- _Helper.with_count(lambda i, _: i in index_set)
+ _Helpers.auto_count(lambda i, _: i in index_set)
)
- def remove_values(self, values: Iterable[Any]) -> _Wrapper[_T]:
+ def remove_values(self, values: Iterable[Any]) -> CruIterator[_T]:
value_set = set(values)
return self.filter(lambda v: v not in value_set)
@@ -450,9 +437,8 @@ class _Wrapper(Generic[_T]):
class CruIterable:
Generic: TypeAlias = _Generic
- Helper: TypeAlias = _Helper
- Creators: TypeAlias = _Creators
- Wrapper: TypeAlias = _Wrapper
+ Iterator: TypeAlias = CruIterator
+ Helpers: TypeAlias = _Helpers
-CRU.add_objects(CruIterable, _Wrapper)
+CRU.add_objects(CruIterable, CruIterator)
diff --git a/tools/cru-py/cru/_list.py b/tools/cru-py/cru/_list.py
index 45122bf..c65c793 100644
--- a/tools/cru-py/cru/_list.py
+++ b/tools/cru-py/cru/_list.py
@@ -1,17 +1,31 @@
from __future__ import annotations
-from collections.abc import Callable
-from typing import Generic, Iterable, Self, TypeAlias, TypeVar
-from ._iter import CruIterable
+from collections.abc import Callable, Iterator
+from typing import Any, Generic, Iterable, TypeAlias, TypeVar
+
+from ._base import CruInternalError
+from ._iter import CruIterator
+from ._const import CruNotFound
_T = TypeVar("_T")
+_O = TypeVar("_O")
+
+
+class CruListEdit(CruIterator[_T]):
+ def __init__(self, iterable: Iterable[_T], _list: CruList[Any]) -> None:
+ super().__init__(iterable)
+ self._list = _list
+
+ def create_me(self, iterable: Iterable[_O]) -> CruListEdit[_O]:
+ return CruListEdit(iterable, self._list)
+ @property
+ def list(self) -> CruList[Any]:
+ return self._list
-class CruListEdit(CruIterable.Wrapper[_T]):
- def done(self) -> CruList[_T]:
- l: CruList[_T] = self.my_attr["list"]
- l.reset(self)
- return l
+ def done(self) -> CruList[Any]:
+ self._list.reset(self)
+ return self._list
class CruList(list[_T]):
@@ -22,11 +36,8 @@ class CruList(list[_T]):
self.extend(new_values)
return self
- def as_iterable_wrapper(self) -> CruIterable.Wrapper[_T]:
- return CruIterable.Wrapper(self)
-
- def as_edit_iterable_wrapper(self) -> CruListEdit[_T]:
- return CruListEdit(self, attr={"list": self})
+ def as_cru_iterator(self) -> CruIterator[_T]:
+ return CruIterator(self)
@staticmethod
def make(maybe_list: Iterable[_T] | _T | None) -> CruList[_T]:
@@ -65,62 +76,70 @@ class CruUniqueKeyList(Generic[_T, _K]):
def validate_self(self):
keys = self._list.transform(self._key_getter)
if len(keys) != len(set(keys)):
- raise ValueError("Duplicate keys!")
-
- # TODO: Continue here!
- def get_or(self, key: K, fallback: Any = CRU_NOT_FOUND) -> _V | Any:
- r = self._l.find_if(lambda i: k == self._key_getter(i))
- return r if r is not CRU_NOT_FOUND else fallback
-
- def get(self, k: K) -> _V:
- 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._key_getter(i) in k)
-
- def try_remove(self, k: K) -> bool:
- i = self._l.find_index_if(lambda v: k == self._key_getter(v))
- if i is CRU_NOT_FOUND:
+ raise CruInternalError("Duplicate keys!")
+
+ def get_or(
+ self, key: _K, fallback: _O | CruNotFound = CruNotFound.VALUE
+ ) -> _T | _O | CruNotFound:
+ return (
+ self._list.as_cru_iterator()
+ .filter(lambda v: key == self._key_getter(v))
+ .first_or(fallback)
+ )
+
+ def get(self, key: _K) -> _T:
+ value = self.get_or(key)
+ if value is CruNotFound:
+ raise KeyError(f"Key {key} not found!")
+ return value # type: ignore
+
+ def has_key(self, key: _K) -> bool:
+ return self.get_or(key) != CruNotFound.VALUE
+
+ def try_remove(self, key: _K) -> bool:
+ value = self.get_or(key)
+ if value is CruNotFound.VALUE:
return False
- self._l.remove_by_indices(i)
+ self._list.remove(value)
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 remove(self, key: _K, allow_absence: bool = False) -> None:
+ if not self.try_remove(key) and not allow_absence:
+ raise KeyError(f"Key {key} not found!")
- def add(self, v: _V, /, replace: bool = False) -> None:
- if self.has_key(self._key_getter(v)):
- if replace:
- self.remove(self._key_getter(v))
- else:
- raise ValueError(f"Key {self._key_getter(v)} already exists!")
+ def add(self, value: _T, /, replace: bool = False) -> None:
+ v = self.get_or(self._key_getter(value))
+ if v is not CruNotFound.VALUE:
+ if not replace:
+ raise KeyError(f"Key {self._key_getter(v)} already exists!")
+ self._list.remove(v)
if self._before_add is not None:
- v = self._before_add(v)
- self._l.append(v)
-
- def set(self, v: _V) -> None:
- self.add(v, True)
-
- def extend(self, l: Iterable[_V], /, replace: bool = False) -> None:
- if not replace and self.has_any_key([self._key_getter(i) for i in l]):
- raise ValueError("Keys already exists!")
+ value = self._before_add(value)
+ self._list.append(value)
+
+ def set(self, value: _T) -> None:
+ self.add(value, True)
+
+ def extend(self, iterable: Iterable[_T], /, replace: bool = False) -> None:
+ values = list(iterable)
+ to_remove = []
+ for value in values:
+ v = self.get_or(self._key_getter(value))
+ if v is not CruNotFound.VALUE:
+ if not replace:
+ raise KeyError(f"Key {self._key_getter(v)} already exists!")
+ to_remove.append(v)
+ for value in to_remove:
+ self._list.remove(value)
if self._before_add is not None:
- l = [self._before_add(i) for i in l]
- keys = [self._key_getter(i) for i in l]
- self._l.remove_all_if(lambda i: self._key_getter(i) in keys).extend(l)
+ values = [self._before_add(value) for value in values]
+ self._list.extend(values)
def clear(self) -> None:
- self._l.clear()
+ self._list.reset([])
- def __iter__(self):
- return iter(self._l)
+ def __iter__(self) -> Iterator[_T]:
+ return iter(self._list)
- def __len__(self):
- return len(self._l)
+ def __len__(self) -> int:
+ return len(self._list)
diff --git a/tools/cru-py/cru/_type.py b/tools/cru-py/cru/_type.py
index c60a913..0fd86a4 100644
--- a/tools/cru-py/cru/_type.py
+++ b/tools/cru-py/cru/_type.py
@@ -1,61 +1,52 @@
-from types import NoneType
+from collections.abc import Iterable
from typing import Any
-DEFAULT_NONE_ERR = ValueError
+from ._base import CruException, CruInternalError
+from ._iter import CruIterator
+
+
+class CruTypeCheckError(CruException):
+ pass
+
+
DEFAULT_NONE_ERR_MSG = "None is not allowed here."
-DEFAULT_TYPE_ERR = ValueError
-DEFAULT_TYPE_ERR_MSG = "Type of object is not allowed here."
+DEFAULT_TYPE_ERR_MSG = "Object of this type is not allowed here."
-class TypeSet:
+class CruTypeSet(set[type]):
def __init__(self, *types: 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)
+ type_set = CruIterator(types).filter(lambda t: t is not None).to_set()
+ if not CruIterator(type_set).all(lambda t: isinstance(t, type)):
+ raise CruInternalError("TypeSet can only contain type.")
+ super().__init__(type_set)
def check_value(
self,
- v: Any,
+ value: Any,
/,
- allow_none: bool,
+ allow_none: bool = False,
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 value is None:
if allow_none:
return
else:
- raise none_err(none_err_msg)
+ raise CruTypeCheckError(DEFAULT_NONE_ERR_MSG)
if len(self) == 0 and empty_allow_all:
return
- if type(v) not in self:
- raise type_err(type_err_msg)
+ if not CruIterator(self).any(lambda t: isinstance(value, t)):
+ raise CruTypeCheckError(DEFAULT_TYPE_ERR_MSG)
def check_value_list(
self,
- l: CanBeList,
+ values: Iterable[Any],
/,
- allow_none: bool,
+ allow_none: bool = False,
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:
+ for value in values:
self.check_value(
- v,
+ value,
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,
)
diff --git a/tools/cru-py/cru/attr.py b/tools/cru-py/cru/attr.py
index a52585a..82f1eba 100644
--- a/tools/cru-py/cru/attr.py
+++ b/tools/cru-py/cru/attr.py
@@ -1,22 +1,27 @@
+from __future__ import annotations
+
import copy
from collections.abc import Callable, Iterable
from dataclasses import dataclass, field
-from typing import Any, ClassVar
+from typing import Any
-from ._util import CanBeList, TypeSet, F, L, WF, CruUniqueKeyInplaceList, CRU_NOT_FOUND, CRU_USE_DEFAULT, \
- CRU_DONT_CHANGE, CRU_PLACEHOLDER
+from ._list import CruUniqueKeyList
+from ._type import CruTypeSet
+from ._const import CruNotFound, CruUseDefault, CruDontChange
+from ._iter import CruIterator
@dataclass
class CruAttr:
- USE_DEFAULT: ClassVar = CRU_USE_DEFAULT
name: str
value: Any
- description: str
+ description: str | None
@staticmethod
- def make(name: str, value: Any = USE_DEFAULT, description: str | None = None) -> "CruAttr":
+ def make(
+ name: str, value: Any = CruUseDefault.VALUE, description: str | None = None
+ ) -> CruAttr:
return CruAttr(name, value, description)
@@ -33,8 +38,14 @@ class CruAttrDef:
transformer: CruAttrTransformer
validator: CruAttrValidator
- def __init__(self, name: str, description: str, default_factory: CruAttrDefaultFactory,
- transformer: CruAttrTransformer, validator: CruAttrValidator) -> 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
@@ -52,7 +63,9 @@ class CruAttrDef:
if self.validator is not None:
self.validator(value, self)
- def transform_and_validate(self, value: Any, /, 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
@@ -68,7 +81,7 @@ class CruAttrDef:
elif attr.name != self.name:
raise ValueError(f"Attr name is not match: {attr.name} != {self.name}")
- if attr.value is CruAttr.USE_DEFAULT:
+ if attr.value is CruUseDefault.VALUE:
attr.value = self.make_default_value()
else:
attr.value = self.transform_and_validate(attr.value)
@@ -78,22 +91,26 @@ class CruAttrDef:
return attr
- 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
+ def make(
+ self, value: Any = CruUseDefault.VALUE, description: None | str = None
+ ) -> CruAttr:
+ value = self.make_default_value() if value is CruUseDefault.VALUE else value
value = self.transform_and_validate(value)
- return CruAttr(self.name, value,
- description if description is not None else self.description)
+ return CruAttr(
+ self.name,
+ value,
+ description if description is not None else self.description,
+ )
@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=CruAttr.USE_DEFAULT)
+ types: list[type] | None = field(default=None)
+ allow_none: bool = field(default=False)
+ default: Any = field(default=CruUseDefault.VALUE)
default_factory: CruAttrDefaultFactory | None = field(default=None)
auto_list: bool = field(default=False)
transformers: list[CruAttrTransformer] = field(default_factory=list)
@@ -108,104 +125,106 @@ class CruAttrDefBuilder:
self.name = name
self.description = description
- def auto_adjust_default(self) -> "CruAttrDefBuilder":
- if self.default is not CruAttr.USE_DEFAULT and self.default is not None:
- return self
- if self.allow_none and self.default is CruAttr.USE_DEFAULT:
+ def auto_adjust_default(self) -> None:
+ if self.default is not CruUseDefault.VALUE and self.default is not None:
+ return
+ if self.allow_none and self.default is CruUseDefault.VALUE:
self.default = None
if not self.allow_none and self.default is None:
- self.default = CruAttr.USE_DEFAULT
+ self.default = CruUseDefault.VALUE
if self.auto_list and not self.allow_none:
self.default = []
- def with_name(self, name: str) -> "CruAttrDefBuilder":
- self.name = name
+ def with_name(self, name: str | CruDontChange) -> CruAttrDefBuilder:
+ if name is not CruDontChange.VALUE:
+ self.name = name
return self
- def with_description(self, default_description: str) -> "CruAttrDefBuilder":
- self.description = default_description
+ def with_description(
+ self, default_description: str | CruDontChange
+ ) -> CruAttrDefBuilder:
+ if default_description is not CruDontChange.VALUE:
+ self.description = default_description
return self
- 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:
+ def with_default(self, default: Any) -> CruAttrDefBuilder:
+ if default is not CruDontChange.VALUE:
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
+ def with_default_factory(
+ self,
+ default_factory: CruAttrDefaultFactory | CruDontChange,
+ ) -> CruAttrDefBuilder:
+ if default_factory is not CruDontChange.VALUE:
+ self.default_factory = default_factory
return self
- def with_default_factory(self, default_factory: CruAttrDefaultFactory) -> "CruAttrDefBuilder":
- self.default_factory = default_factory
+ def with_types(
+ self,
+ types: Iterable[type] | None | CruDontChange,
+ ) -> CruAttrDefBuilder:
+ if types is not CruDontChange.VALUE:
+ self.types = None if types is None else list(types)
return self
- def with_types(self, allowed_types: CanBeList[type], default: Any = DONT_CHANGE) -> "CruAttrDefBuilder":
- self.types = allowed_types
- if default is not CruAttrDefBuilder.DONT_CHANGE:
- self.default = default
+ def with_allow_none(self, allow_none: bool | CruDontChange) -> CruAttrDefBuilder:
+ if allow_none is not CruDontChange.VALUE:
+ self.allow_none = allow_none
return self
- def with_allow_none(self, allow_none: bool, default: Any = DONT_CHANGE) -> "CruAttrDefBuilder":
- self.allow_none = allow_none
- if default is not CruAttrDefBuilder.DONT_CHANGE:
- self.default = default
- return self
-
- def with_optional(self, default: Any = DONT_CHANGE) -> "CruAttrDefBuilder":
- return self.with_allow_none(True, default)
-
- def with_required(self, default: Any = DONT_CHANGE) -> "CruAttrDefBuilder":
- return self.with_allow_none(False, default)
-
- 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):
+ def with_auto_list(
+ self, auto_list: bool | CruDontChange = True
+ ) -> CruAttrDefBuilder:
+ if auto_list is not CruDontChange.VALUE:
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, "CruAttrDef"], Any] | None) -> "CruAttrDefBuilder":
- if transformer is not None:
- self.transformers.append(transformer)
+ def with_constraint(
+ self,
+ /,
+ allow_none: bool | CruDontChange = CruDontChange.VALUE,
+ types: Iterable[type] | None | CruDontChange = CruDontChange.VALUE,
+ default: Any = CruDontChange.VALUE,
+ default_factory: CruAttrDefaultFactory | CruDontChange = CruDontChange.VALUE,
+ auto_list: bool | CruDontChange = CruDontChange.VALUE,
+ ) -> CruAttrDefBuilder:
+ return (
+ self.with_allow_none(allow_none)
+ .with_types(types)
+ .with_default(default)
+ .with_default_factory(default_factory)
+ .with_auto_list(auto_list)
+ )
+
+ def add_transformer(self, transformer: CruAttrTransformer) -> CruAttrDefBuilder:
+ self.transformers.append(transformer)
return self
- def clear_transformers(self) -> "CruAttrDefBuilder":
+ def clear_transformers(self) -> CruAttrDefBuilder:
self.transformers.clear()
return self
- def add_validator(self, validator: Callable[[Any, "CruAttrDef"], None] | None) -> "CruAttrDefBuilder":
- if validator is not None:
- self.validators.append(validator)
+ def add_validator(self, validator: CruAttrValidator) -> CruAttrDefBuilder:
+ self.validators.append(validator)
return self
- def clear_validators(self) -> "CruAttrDefBuilder":
+ 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
+ def with_override_transformer(
+ self, override_transformer: CruAttrTransformer | None | CruDontChange
+ ) -> CruAttrDefBuilder:
+ if override_transformer is not CruDontChange.VALUE:
+ self.override_transformer = override_transformer
return self
- def with_override_validator(self, override_validator: Callable[[Any], None] | None) -> "CruAttrDefBuilder":
- self.override_validator = override_validator
+ def with_override_validator(
+ self, override_validator: CruAttrValidator | None | CruDontChange
+ ) -> CruAttrDefBuilder:
+ if override_validator is not CruDontChange.VALUE:
+ self.override_validator = override_validator
return self
def is_valid(self) -> tuple[bool, str]:
@@ -213,50 +232,72 @@ class CruAttrDefBuilder:
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:
+ 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!"
+ return True, ""
@staticmethod
- def _build(b: "CruAttrDefBuilder", auto_adjust_default: bool = True) -> CruAttrDef:
+ def _build(
+ builder: 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, 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)
- v = v.transform(transformer)
+ builder.auto_adjust_default()
+
+ valid, err = builder.is_valid()
+ if not valid:
+ raise ValueError(err)
+
+ def composed_transformer(value: Any, attr_def: CruAttrDef) -> Any:
+ def transform_value(single_value: Any) -> Any:
+ for transformer in builder.transformers:
+ single_value = transformer(single_value, attr_def)
+ return single_value
+
+ if builder.auto_list:
+ if not isinstance(value, list):
+ value = [value]
+ value = CruIterator(value).transform(transform_value).to_list()
+
else:
- v = transformer(v)
- return v
-
- type_set = TypeSet(b.types)
-
- 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)
- L(v).foreach(validator)
+ value = transform_value(value)
+ return value
+
+ type_set = None if builder.types is None else CruTypeSet(*builder.types)
+
+ def composed_validator(value: Any, attr_def: CruAttrDef):
+ def validate_value(single_value: Any) -> None:
+ if type_set is not None:
+ type_set.check_value(single_value, allow_none=builder.allow_none)
+ for validator in builder.validators:
+ validator(single_value, attr_def)
+
+ if builder.auto_list:
+ CruIterator(value).foreach(validate_value)
else:
- type_set.check_value(v, allow_none=b.allow_none)
- validator(v)
+ validate_value(value)
- real_transformer = b.override_transformer or composed_transformer
- real_validator = b.override_validator or composed_validator
+ real_transformer = builder.override_transformer or composed_transformer
+ real_validator = builder.override_validator or composed_validator
- default_factory = b.default_factory
+ default_factory = builder.default_factory
if default_factory is None:
- default_factory = lambda _d: copy.deepcopy(b.default)
- d = CruAttrDef(b.name, b.description, default_factory, real_transformer, real_validator)
- if b.build_hook: b.build_hook(d)
+ def default_factory(_d):
+ return copy.deepcopy(builder.default)
+
+ d = CruAttrDef(
+ builder.name,
+ builder.description,
+ default_factory,
+ real_transformer,
+ real_validator,
+ )
+ if builder.build_hook:
+ builder.build_hook(d)
return d
def build(self, auto_adjust_default=True) -> CruAttrDef:
@@ -265,7 +306,7 @@ class CruAttrDefBuilder:
return CruAttrDefBuilder._build(c, auto_adjust_default)
-class CruAttrDefRegistry(CruUniqueKeyInplaceList[CruAttrDef, str]):
+class CruAttrDefRegistry(CruUniqueKeyList[CruAttrDef, str]):
def __init__(self) -> None:
super().__init__(lambda d: d.name)
@@ -280,7 +321,7 @@ class CruAttrDefRegistry(CruUniqueKeyInplaceList[CruAttrDef, str]):
return d.adopt(attr)
-class CruAttrTable(CruUniqueKeyInplaceList[CruAttr, str]):
+class CruAttrTable(CruUniqueKeyList[CruAttr, str]):
def __init__(self, registry: CruAttrDefRegistry) -> None:
self._registry: CruAttrDefRegistry = registry
super().__init__(lambda a: a.name, before_add=registry.adopt)
@@ -289,9 +330,9 @@ class CruAttrTable(CruUniqueKeyInplaceList[CruAttr, str]):
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:
+ def get_value_or(self, name: str, fallback: Any = CruNotFound.VALUE) -> Any:
+ a = self.get_or(name, CruNotFound.VALUE)
+ if a is CruNotFound.VALUE:
return fallback
return a.value
@@ -299,43 +340,25 @@ class CruAttrTable(CruUniqueKeyInplaceList[CruAttr, str]):
a = self.get(name)
return a.value
- def make_attr(self, name: str, value: Any = CruAttr.USE_DEFAULT, /, description: str | None = None) -> CruAttr:
+ def make_attr(
+ self,
+ name: str,
+ value: Any = CruUseDefault.VALUE,
+ /,
+ description: str | None = None,
+ ) -> CruAttr:
d = self._registry.get(name)
return d.make(value, description or d.description)
- def add_value(self, name: str, value: Any = CruAttr.USE_DEFAULT, /, description: str | None = None, *,
- replace: bool = False) -> CruAttr:
+ def add_value(
+ self,
+ name: str,
+ value: Any = CruUseDefault.VALUE,
+ /,
+ description: str | None = None,
+ *,
+ replace: bool = False,
+ ) -> CruAttr:
attr = self.make_attr(name, value, description)
self.add(attr, replace)
return attr
-
- 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 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
-
- 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
-
- 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
-
- raise TypeError(f"Unsupported type: {type(a)}")
diff --git a/tools/cru-py/cru/config.py b/tools/cru-py/cru/config.py
index b0c83d5..843f30a 100644
--- a/tools/cru-py/cru/config.py
+++ b/tools/cru-py/cru/config.py
@@ -1,6 +1,6 @@
from typing import Any, TypeVar, Generic
-from .excp import CruInternalLogicError
+from .error import CruInternalLogicError
from .value import ValueType, ValueGenerator, ValidationError
T = TypeVar("T")
diff --git a/tools/cru-py/cru/error.py b/tools/cru-py/cru/error.py
new file mode 100644
index 0000000..95edbd3
--- /dev/null
+++ b/tools/cru-py/cru/error.py
@@ -0,0 +1,27 @@
+from typing import NoReturn
+
+
+class CruException(Exception):
+ """Base exception class of all exceptions in cru."""
+
+
+class CruUnreachableError(CruException):
+ """Raised when a code path is unreachable."""
+
+
+def cru_unreachable() -> NoReturn:
+ raise CruUnreachableError()
+
+
+class CruInternalError(CruException):
+ """Raised when an internal logic error occurs."""
+
+
+class UserFriendlyException(CruException):
+ def __init__(self, message: str, user_message: str, *args, **kwargs) -> None:
+ super().__init__(message, *args, **kwargs)
+ self._user_message = user_message
+
+ @property
+ def user_message(self) -> str:
+ return self._user_message
diff --git a/tools/cru-py/cru/excp.py b/tools/cru-py/cru/excp.py
deleted file mode 100644
index 9ea204e..0000000
--- a/tools/cru-py/cru/excp.py
+++ /dev/null
@@ -1,116 +0,0 @@
-from typing import Any
-
-from .attr import CruAttrDefRegistry, CruAttr, CruAttrTable
-from ._util import CRU_NOT_FOUND, CruList, CRU_USE_DEFAULT
-
-CRU_EXCEPTION_ATTR_DEF_REGISTRY = CruAttrDefRegistry()
-
-
-class CruException(Exception):
- ATTR_REGISTRY = CRU_EXCEPTION_ATTR_DEF_REGISTRY
-
- MESSAGE_KEY = "message"
- INNER_KEY = "inner"
- INTERNAL_KEY = "internal"
- NAME_KEY = "name"
- VALUE_KEY = "value"
- PATH_KEY = "path"
- TYPE_KEY = "type_"
-
- 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()
- 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()
-
- def __init__(self, message: str, *args,
- 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: CruAttrTable = CruAttrTable(self.ATTR_REGISTRY)
-
- self._attrs.add_value(CruException.MESSAGE_KEY, message)
- if init_attrs is not None:
- self._attrs.extend_with(init_attrs, True)
- if attrs is not None:
- 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.get_attr(CruException.MESSAGE_KEY)
-
- @property
- def internal(self) -> bool:
- return self.get_attr(CruException.INTERNAL_KEY)
-
- @property
- def inner(self) -> list[Exception]:
- return self.get_attr(CruException.INNER_KEY)
-
- @property
- def name(self) -> str | None:
- return self.get_attr(CruException.NAME_KEY)
-
- @property
- def value(self) -> Any | None:
- return self.get_attr(CruException.VALUE_KEY)
-
- @property
- def path(self) -> str | None:
- return self.get_attr(CruException.PATH_KEY)
-
- @property
- 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_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, des: type = CruList) -> list[Any]:
- l = []
- self._get_attr_list_recursive(name, 0, max_depth, l)
- return des(l)
-
-
-class CruInternalLogicError(CruException):
- def __init__(self, message: str, *args, **kwargs) -> None:
- super().__init__(message, *args, internal=True, **kwargs)
-
-
-class UserFriendlyException(CruException):
- USER_MESSAGE_KEY = "user_message"
-
- 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 = 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.get_attr(UserFriendlyException.USER_MESSAGE_KEY)
diff --git a/tools/cru-py/cru/paths.py b/tools/cru-py/cru/paths.py
index 937d2ac..cdd97fe 100644
--- a/tools/cru-py/cru/paths.py
+++ b/tools/cru-py/cru/paths.py
@@ -1,7 +1,7 @@
import os
from pathlib import Path
-from .excp import CruException
+from .error import CruException
class ApplicationPathError(CruException):
@@ -39,7 +39,8 @@ class ApplicationPath:
if not self.check_parents(must_exist):
return False
if not self.path.exists():
- if not must_exist: return False
+ if not must_exist:
+ return False
raise ApplicationPathError("Mot exist.", self.path)
if self.is_dir:
if not self.path.is_dir():
diff --git a/tools/cru-py/cru/value.py b/tools/cru-py/cru/value.py
index 7b716f0..189f44f 100644
--- a/tools/cru-py/cru/value.py
+++ b/tools/cru-py/cru/value.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
import random
import secrets
import string
@@ -6,7 +8,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
+from .error import CruInternalError, CruException
def _str_case_in(s: str, case: bool, l: list[str]) -> bool:
@@ -16,86 +18,77 @@ def _str_case_in(s: str, case: bool, l: list[str]) -> bool:
return s.lower() in [s.lower() for s in l]
-_ValueTypeForward = type["ValueType"]
-
T = TypeVar("T")
-class _ValueErrorMixin:
- VALUE_TYPE_KEY = "value_type"
-
- CruException.ATTR_REGISTRY.make_builder(
- VALUE_TYPE_KEY,
- "The type of the value that causes the exception."
- ).build()
-
+class CruValueError(CruException):
+ def __init__(
+ self,
+ message: str,
+ value: Any,
+ value_type: ValueType | None,
+ *args,
+ **kwargs,
+ ):
+ super().__init__(
+ message,
+ *args,
+ **kwargs,
+ )
+ self._value = value
+ self._value_type = value_type
-class ValidationError(CruException, _ValueErrorMixin):
- def __init__(self, message: str, value: Any, value_type: _ValueTypeForward[T] | None, *args, **kwargs):
- super().__init__(message, *args, value=value, type_=value_type.type, init_attrs={
- ValidationError.VALUE_TYPE_KEY: value_type,
- }, **kwargs)
+ @property
+ def value(self) -> Any:
+ return self._value
@property
- def value_type(self) -> _ValueTypeForward[T] | None:
- return self.get_attr(ValidationError.VALUE_TYPE_KEY)
+ def value_type(self) -> ValueType | None:
+ return self._value_type
-class ValueStringConvertionError(CruException, _ValueErrorMixin):
- def __init__(self, message: str, value: Any, value_type: _ValueTypeForward[T] | None, *args,
- **kwargs):
- super().__init__(message, *args, value=value, type_=value_type.type, init_attrs={
- ValueStringConvertionError.VALUE_TYPE_KEY: value_type,
- }, **kwargs)
+class CruValueValidationError(CruValueError):
+ pass
- @property
- def value_type(self) -> _ValueTypeForward[T] | None:
- return self.get_attr(ValidationError.VALUE_TYPE_KEY)
+class CruValueStringConversionError(CruValueError):
+ pass
-class ValueType(Generic[T], metaclass=ABCMeta):
- @staticmethod
- def case_sensitive_to_str(case_sensitive: bool) -> str:
- return f"case-{'' if case_sensitive else 'in'}sensitive"
- def __init__(self, name: str) -> None:
+# TODO: Continue here tomorrow!
+class ValueType(Generic[T], metaclass=ABCMeta):
+ def __init__(self, name: str, _type: type[T]) -> None:
self._name = name
- self._type = type("T")
+ self._type = _type
@property
def name(self) -> str:
return self._name
@property
- def type(self) -> type:
+ def type(self) -> type[T]:
return self._type
- def is_instance_of_value_type(self, value: Any) -> bool:
+ def check_value_type(self, value: Any) -> bool:
return isinstance(value, self.type)
- def _do_check_value(self, value: Any) -> tuple[True, T] | tuple[False, None | str]:
- return True, value
+ def _do_check_value(self, value: Any) -> T:
+ return value
def check_value(self, value: Any) -> T:
if not isinstance(value, self.type):
- raise ValidationError("Value type is wrong.", value, self)
- ok, v_or_err = self._do_check_value(value)
- if ok:
- return v_or_err
- else:
- raise ValidationError(v_or_err or "Value is not valid.", value, self)
+ raise CruValueValidationError("Value type is wrong.", value, self)
+ return self._do_check_value(value)
- @abstractmethod
def _do_check_str_format(self, s: str) -> bool | tuple[bool, str]:
- """
- Return None for no error. Otherwise, return error message.
- """
raise NotImplementedError()
def check_str_format(self, s: str) -> None:
ok, err = self._do_check_str_format(s)
- if ok is None: raise CruInternalLogicError("_do_check_str_format should not return None.")
- if ok: return
+ if ok is None:
+ raise CruInternalLogicError("_do_check_str_format should not return None.")
+ if ok:
+ return
if err is None:
err = "Invalid value str format."
raise ValueStringConvertionError(err, s, value_type=self)
@@ -123,8 +116,9 @@ class ValueType(Generic[T], metaclass=ABCMeta):
if isinstance(value_or_str, str):
return self.convert_str_to_value(value_or_str)
else:
- raise ValidationError("Value is not valid and is not a str.", value_or_str, self,
- inner=e)
+ raise ValidationError(
+ "Value is not valid and is not a str.", value_or_str, self, inner=e
+ )
class TextValueType(ValueType[str]):
@@ -182,12 +176,21 @@ class BooleanValueType(ValueType[bool]):
DEFAULT_TRUE_LIST: ClassVar[list[str]] = ["true", "yes", "y", "on", "1"]
DEFAULT_FALSE_LIST: ClassVar[list[str]] = ["false", "no", "n", "off", "0"]
- def __init__(self, *, case_sensitive=False, true_list: None | list[str] = None,
- false_list: None | list[str] = None) -> None:
+ def __init__(
+ self,
+ *,
+ case_sensitive=False,
+ true_list: None | list[str] = None,
+ false_list: None | list[str] = None,
+ ) -> None:
super().__init__("boolean")
self._case_sensitive = case_sensitive
- self._valid_true_strs: list[str] = true_list or BooleanValueType.DEFAULT_TRUE_LIST
- self._valid_false_strs: list[str] = false_list or BooleanValueType.DEFAULT_FALSE_LIST
+ self._valid_true_strs: list[str] = (
+ true_list or BooleanValueType.DEFAULT_TRUE_LIST
+ )
+ self._valid_false_strs: list[str] = (
+ false_list or BooleanValueType.DEFAULT_FALSE_LIST
+ )
@property
def case_sensitive(self) -> bool:
@@ -206,8 +209,12 @@ class BooleanValueType(ValueType[bool]):
return self._valid_true_strs + self._valid_false_strs
def _do_check_str_format(self, s):
- if _str_case_in(s, self.case_sensitive, self.valid_boolean_strs): return True
- return False, f"Not a valid boolean string ({ValueType.case_sensitive_to_str(self.case_sensitive)}). Valid string of true: {' '.join(self._valid_true_strs)}. Valid string of false: {' '.join(self._valid_false_strs)}. All is case insensitive."
+ if _str_case_in(s, self.case_sensitive, self.valid_boolean_strs):
+ return True
+ return (
+ False,
+ f"Not a valid boolean string ({ValueType.case_sensitive_to_str(self.case_sensitive)}). Valid string of true: {' '.join(self._valid_true_strs)}. Valid string of false: {' '.join(self._valid_false_strs)}. All is case insensitive.",
+ )
def _do_convert_value_to_str(self, value):
return "True" if value else "False"
@@ -218,8 +225,8 @@ class BooleanValueType(ValueType[bool]):
class EnumValueType(ValueType[str]):
def __init__(self, valid_values: list[str], /, case_sensitive=False) -> None:
- s = ' | '.join([f'"{v}"' for v in valid_values])
- self._valid_value_str = f'[ {s} ]'
+ s = " | ".join([f'"{v}"' for v in valid_values])
+ self._valid_value_str = f"[ {s} ]"
super().__init__(f"enum{self._valid_value_str}")
self._case_sensitive = case_sensitive
self._valid_values = valid_values
@@ -237,8 +244,12 @@ class EnumValueType(ValueType[str]):
return ok, (value if ok else err)
def _do_check_str_format(self, s):
- if _str_case_in(s, self.case_sensitive, self.valid_values): return True
- return False, f"Value is not in valid values ({ValueType.case_sensitive_to_str(self.case_sensitive)}): {self._valid_value_str}"
+ if _str_case_in(s, self.case_sensitive, self.valid_values):
+ return True
+ return (
+ False,
+ f"Value is not in valid values ({ValueType.case_sensitive_to_str(self.case_sensitive)}): {self._valid_value_str}",
+ )
def _do_convert_value_to_str(self, value):
return value
@@ -251,13 +262,15 @@ TEXT_VALUE_TYPE = TextValueType()
INTEGER_VALUE_TYPE = IntegerValueType()
BOOLEAN_VALUE_TYPE = BooleanValueType()
-P = ParamSpec('P')
+P = ParamSpec("P")
class ValueGenerator(Generic[T, P]):
INTERACTIVE_KEY: ClassVar[Literal["interactive"]] = "interactive"
- def __init__(self, f: Callable[P, T], /, attributes: None | Mapping[str, Any] = None) -> None:
+ def __init__(
+ self, f: Callable[P, T], /, attributes: None | Mapping[str, Any] = None
+ ) -> None:
self._f = f
self._attributes = attributes or {}
@@ -280,9 +293,15 @@ class ValueGenerator(Generic[T, P]):
return self._attributes.get(ValueGenerator.INTERACTIVE_KEY, False)
@staticmethod
- def create_interactive(f: Callable[P, T], interactive: bool = True, /,
- attributes: None | Mapping[str, Any] = None) -> "ValueGenerator[T, P]":
- return ValueGenerator(f, dict({ValueGenerator.INTERACTIVE_KEY: interactive}, **(attributes or {})))
+ def create_interactive(
+ f: Callable[P, T],
+ interactive: bool = True,
+ /,
+ attributes: None | Mapping[str, Any] = None,
+ ) -> "ValueGenerator[T, P]":
+ return ValueGenerator(
+ f, dict({ValueGenerator.INTERACTIVE_KEY: interactive}, **(attributes or {}))
+ )
class UuidValueGenerator(ValueGenerator[str, []]):
@@ -297,13 +316,15 @@ class RandomStringValueGenerator(ValueGenerator[str, []]):
def generate_random_string():
characters = string.ascii_letters + string.digits
- random_string = ''.join(random_choice(characters) for _ in range(length))
+ random_string = "".join(random_choice(characters) for _ in range(length))
return random_string
return generate_random_string
def __init__(self, length: int, secure: bool) -> None:
- super().__init__(RandomStringValueGenerator._create_generate_ramdom_func(length, secure))
+ super().__init__(
+ RandomStringValueGenerator._create_generate_ramdom_func(length, secure)
+ )
UUID_VALUE_GENERATOR = UuidValueGenerator()
generated by cgit v1.2.3 (git 2.39.1) at 2025-04-08 10:44:22 +0000