sh-edraft.de/sh_cpl
2
1
Fork 0
Code Issues 4 Pull Requests 1 Actions Packages Projects 1 Releases 12 Wiki Activity

Refactored cpl-query Iterable & Enumerable #129

Browse Source
This commit is contained in:
Sven Heidemann
2022-12-01 21:09:39 +01:00
parent e8ae635c88
commit f0f79e7e3b
19 changed files with 390 additions and 916 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/cpl_query/enumerable/__init__.py
View File

@@ -19,12 +19,9 @@ __version__ = '2022.10.0.post2'
from collections import namedtuple
# imports:
from .enumerable import Enumerable
from .enumerable_abc import EnumerableABC
from .ordered_enumerable import OrderedEnumerable
from .ordered_enumerable_abc import OrderedEnumerableABC
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='10', micro='0.post2')

296
src/cpl_query/enumerable/enumerable.py
View File

@@ -1,9 +1,4 @@
from typing import Union, Callable, Optional
from cpl_query._helper import is_number
from cpl_query.enumerable.enumerable_abc import EnumerableABC
from cpl_query.enumerable.ordered_enumerable_abc import OrderedEnumerableABC
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument, InvalidTypeException, IndexOutOfRangeException
def _default_lambda(x: object):
@@ -14,294 +9,5 @@ class Enumerable(EnumerableABC):
r"""Implementation of :class: `cpl_query.enumerable.enumerable_abc.EnumerableABC`
"""
def __init__(self, t: type = None, values: Union[list, iter] = None):
def __init__(self, t: type = None, values: list = None):
EnumerableABC.__init__(self, t, values)
def all(self, _func: Callable = None) -> bool:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
_func = _default_lambda
result = self.where(_func)
return len(result) == len(self)
def any(self, _func: Callable = None) -> bool:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
_func = _default_lambda
result = self.where(_func)
return len(result) > 0
def average(self, _func: Callable = None) -> Union[int, float, complex]:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None and not is_number(self.type):
raise InvalidTypeException()
if _func is None:
_func = _default_lambda
return float(self.sum(_func)) / float(self.count())
def contains(self, _value: object) -> bool:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _value is None:
raise ArgumentNoneException(ExceptionArgument.value)
return self.where(lambda x: x == _value).count() > 0
def count(self, _func: Callable = None) -> int:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
return len(self)
return len(self.where(_func))
def distinct(self, _func: Callable = None) -> EnumerableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
_func = _default_lambda
result = Enumerable()
known_values = []
for element in self:
value = _func(element)
if value in known_values:
continue
known_values.append(value)
result.add(element)
return result
def element_at(self, _index: int) -> any:
self._values.reset()
while _index >= 0:
current = self.next()
if _index == 0:
return current
_index -= 1
def element_at_or_default(self, _index: int) -> any:
try:
return self.element_at(_index)
except IndexOutOfRangeException:
return None
@staticmethod
def empty() -> 'EnumerableABC':
r"""Returns an empty enumerable
Returns
-------
Enumerable object that contains no elements
"""
return Enumerable()
def first(self: EnumerableABC, _func=None) -> any:
if _func is not None:
return self.where(_func).element_at(0)
return self.element_at(0)
def first_or_default(self: EnumerableABC, _func=None) -> Optional[any]:
if _func is not None:
return self.where(_func).element_at_or_default(0)
return self.element_at_or_default(0)
def for_each(self, _func: Callable = None):
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
_func = _default_lambda
for element in self:
_func(element)
def last(self: EnumerableABC) -> any:
return self.element_at(self.count() - 1)
def last_or_default(self: EnumerableABC) -> Optional[any]:
return self.element_at_or_default(self.count() - 1)
def max(self, _func: Callable = None) -> Union[int, float, complex]:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None and not is_number(self.type):
raise InvalidTypeException()
if _func is None:
_func = _default_lambda
return _func(max(self, key=_func))
def median(self, _func=None) -> Union[int, float]:
if _func is None:
_func = _default_lambda
result = self.order_by(_func).select(_func).to_list()
length = len(result)
i = int(length / 2)
return (
result[i]
if length % 2 == 1
else (float(result[i - 1]) + float(result[i])) / float(2)
)
def min(self, _func: Callable = None) -> Union[int, float, complex]:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None and not is_number(self.type):
raise InvalidTypeException()
if _func is None:
_func = _default_lambda
return _func(min(self, key=_func))
def order_by(self, _func: Callable = None) -> OrderedEnumerableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
_func = _default_lambda
from cpl_query.enumerable.ordered_enumerable import OrderedEnumerable
return OrderedEnumerable(self.type, _func, sorted(self, key=_func))
def order_by_descending(self, _func: Callable = None) -> OrderedEnumerableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
_func = _default_lambda
from cpl_query.enumerable.ordered_enumerable import OrderedEnumerable
return OrderedEnumerable(self.type, _func, sorted(self, key=_func, reverse=True))
@staticmethod
def range(start: int, length: int) -> 'EnumerableABC':
return Enumerable(int, range(start, length))
def reverse(self: EnumerableABC) -> EnumerableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
return Enumerable(self.type, list(reversed(self.to_list())))
def select(self, _func: Callable = None) -> EnumerableABC:
if _func is None:
_func = _default_lambda
_l = [_func(_o) for _o in self]
return Enumerable(self._type if len(_l) < 1 else type(_l[0]), _l)
def select_many(self, _func: Callable = None) -> EnumerableABC:
if _func is None:
_func = _default_lambda
# The line below is pain. I don't understand anything of the list comprehension...
# written on 09.11.2022 by Sven Heidemann
_l = [_a for _o in self for _a in _func(_o)]
return Enumerable(self._type if len(_l) < 1 else type(_l[0]), _l)
def single(self: EnumerableABC) -> any:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if len(self) > 1:
raise IndexError('Found more than one element')
elif len(self) == 0:
raise IndexOutOfRangeException(f'{type(self).__name__} is empty')
return self.element_at(0)
def single_or_default(self: EnumerableABC) -> Optional[any]:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if len(self) > 1:
raise IndexError('Found more than one element')
elif len(self) == 0:
return None
return self.element_at(0)
def skip(self, _index: int) -> EnumerableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
_list = self.to_list()
return Enumerable(self.type, _list[_index:])
def skip_last(self, _index: int) -> EnumerableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
index = len(self) - _index
return self.take(len(self) - _index)
def take(self, _index: int) -> EnumerableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
_list = self.to_list()
return Enumerable(self.type, _list[:_index])
def take_last(self, _index: int) -> EnumerableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
_list = self.to_list()
index = len(_list) - _index
return self.skip(index)
def sum(self, _func: Callable = None) -> Union[int, float, complex]:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None and not is_number(self.type):
raise InvalidTypeException()
if _func is None:
_func = _default_lambda
return sum([_func(x) for x in self])
def where(self, _func: Callable = None) -> EnumerableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
return Enumerable(self.type, list(filter(_func, self._values)))

81
src/cpl_query/enumerable/enumerable_abc.py
View File

@@ -1,8 +1,6 @@
from abc import abstractmethod
from typing import Iterable
from cpl_query.base.queryable_abc import QueryableABC
from cpl_query.base.sequence_values import SequenceValues
class EnumerableABC(QueryableABC):
@@ -11,81 +9,15 @@ class EnumerableABC(QueryableABC):
@abstractmethod
def __init__(self, t: type = None, values: list = None):
if t == any or t is None and values is not None:
t = type(values[0])
QueryableABC.__init__(self, t, values)
self._type, self._values, self._remove_error_check = t, SequenceValues(values, t), True
def __len__(self):
return len(self._values)
def __iter__(self):
return iter(self._values)
def next(self):
return next(self._values)
def __next__(self):
return self.next()
def __repr__(self):
return f'<{type(self).__name__} {list(self).__repr__()}>'
@property
def type(self) -> type:
return self._type
self._remove_error_check = True
def set_remove_error_check(self, _value: bool):
r"""Set flag to check if element exists before removing
"""
self._remove_error_check = _value
def add(self, __object: object) -> None:
r"""Adds an element to the enumerable.
"""
if self._type is not None and type(__object) != self._type and not isinstance(type(__object), self._type) and not issubclass(type(__object), self._type):
raise Exception(f'Unexpected type: {type(__object)}\nExpected type: {self._type}')
if len(self) == 0 and self._type is None:
self._type = type(__object)
self._values = SequenceValues([*self._values, __object], self._type)
def clear(self):
r"""Removes all elements
"""
del self._values
self._values = []
def extend(self, __list: Iterable) -> 'EnumerableABC':
r"""Adds elements of given list to enumerable
Parameter
---------
__enumerable: :class: `cpl_query.enumerable.enumerable_abc.EnumerableABC`
index
"""
self._values = SequenceValues([*self._values, *__list], self._type)
return self
def remove(self, __object: object) -> None:
r"""Removes element from list
Parameter
---------
__object: :class:`object`
value
Raises
---------
`Element not found` when element does not exist. Check can be deactivated by calling <enumerable>.set_remove_error_check(False)
"""
if self._remove_error_check and __object not in self._values:
raise Exception('Element not found')
# self._values.remove(__object)
self._values = SequenceValues([x for x in self.to_list() if x != __object], self._type)
def to_iterable(self) -> 'IterableABC':
r"""Converts :class: `cpl_query.enumerable.enumerable_abc.EnumerableABC` to :class: `cpl_query.iterable.iterable_abc.IterableABC`
@@ -95,12 +27,3 @@ class EnumerableABC(QueryableABC):
"""
from cpl_query.iterable.iterable import Iterable
return Iterable(self._type, self.to_list())
def to_list(self) -> list:
r"""Converts :class: `cpl_query.base.sequence_abc.SequenceABC` to :class: `list`
Returns
-------
:class: `list`
"""
return [x for x in self]

36
src/cpl_query/enumerable/ordered_enumerable.py
View File

@@ -1,36 +0,0 @@
from collections.abc import Callable
from typing import Iterable
from cpl_query.enumerable.enumerable import Enumerable
from cpl_query.enumerable.ordered_enumerable_abc import OrderedEnumerableABC
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument
class OrderedEnumerable(Enumerable, OrderedEnumerableABC):
r"""Implementation of :class: `cpl_query.extension.Enumerable` `cpl_query.extension.OrderedEnumerableABC`
"""
def __init__(self, _t: type, _func: Callable = None, _values: Iterable = None):
Enumerable.__init__(self, _t)
OrderedEnumerableABC.__init__(self, _t, _func, _values)
def then_by(self: OrderedEnumerableABC, _func: Callable) -> OrderedEnumerableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
self._funcs.append(_func)
return OrderedEnumerable(self.type, _func, sorted(self, key=lambda *args: [f(*args) for f in self._funcs]))
def then_by_descending(self: OrderedEnumerableABC, _func: Callable) -> OrderedEnumerableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
self._funcs.append(_func)
return OrderedEnumerable(self.type, _func, sorted(self, key=lambda *args: [f(*args) for f in self._funcs], reverse=True))

43
src/cpl_query/enumerable/ordered_enumerable_abc.py
View File

@@ -1,43 +0,0 @@
from abc import abstractmethod
from collections.abc import Callable
from typing import Iterable
from cpl_query.enumerable.enumerable_abc import EnumerableABC
class OrderedEnumerableABC(EnumerableABC):
@abstractmethod
def __init__(self, _t: type, _func: Callable = None, _values: Iterable = None):
EnumerableABC.__init__(self, _t, _values)
self._funcs: list[Callable] = []
if _func is not None:
self._funcs.append(_func)
@abstractmethod
def then_by(self, func: Callable) -> 'OrderedEnumerableABC':
r"""Sorts OrderedList in ascending order by function
Parameter
---------
func: :class:`Callable`
Returns
-------
list of :class:`cpl_query.extension.OrderedEnumerableABC`
"""
pass
@abstractmethod
def then_by_descending(self, func: Callable) -> 'OrderedEnumerableABC':
r"""Sorts OrderedList in descending order by function
Parameter
---------
func: :class:`Callable`
Returns
-------
list of :class:`cpl_query.extension.OrderedEnumerableABC`
"""
pass