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

Release 2022.10 #112

Merged
edraft merged 51 commits from 2022.10 into master 2022-09-30 15:48:19 +02:00
Conversation 0 Commits 51 Files Changed 370 +1791 -699
28 changed files with 1791 additions and 699 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 8dcc5b4011 - Show all commits

4
src/cpl_query/__init__.py
View File

@ -15,7 +15,7 @@ __title__ = 'cpl_query'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2021 - 2022 sh-edraft.de'
__version__ = '2022.10.2'
__version__ = '2022.10.9'
from collections import namedtuple
@ -23,4 +23,4 @@ from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='10', micro='2')
version_info = VersionInfo(major='2022', minor='10', micro='9')

26
src/cpl_query/base/__init__.py Normal file
View File

@ -0,0 +1,26 @@
# -*- coding: utf-8 -*-
"""
cpl-query sh-edraft Common Python library Query
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library Python integrated Queries
:copyright: (c) 2021 - 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_query.base'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2021 - 2022 sh-edraft.de'
__version__ = '2022.10.9'
from collections import namedtuple
# imports:
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='10', micro='9')

188
src/cpl_query/extension/iterable_abc.py → src/cpl_query/base/queryable_abc.py
View File

@ -1,29 +1,11 @@
from abc import ABC, abstractmethod
from typing import Optional, Callable, Union, Iterable
from abc import abstractmethod, ABC
from typing import Optional, Callable, Union
class IterableABC(ABC, list):
r"""ABC to define functions on list
"""
class QueryableABC(ABC):
@abstractmethod
def __init__(self, t: type = None, values: list = None):
list.__init__(self)
if t == any:
t = None
self._type = t
if values is not None:
for value in values:
self.append(value)
@property
def type(self) -> type:
return self._type
@abstractmethod
def all(self, func: Callable) -> bool:
def all(self, _func: Callable = None) -> bool:
r"""Checks if every element of list equals result found by function
Parameter
@ -38,7 +20,7 @@ class IterableABC(ABC, list):
pass
@abstractmethod
def any(self, func: Callable) -> bool:
def any(self, _func: Callable = None) -> bool:
r"""Checks if list contains result found by function
Parameter
@ -52,24 +34,8 @@ class IterableABC(ABC, list):
"""
pass
def append(self, __object: object) -> None:
r"""Adds element to list
Parameter
---------
__object: :class:`object`
value
"""
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)
super().append(__object)
@abstractmethod
def average(self, func: Callable = None) -> Union[int, float, complex]:
def average(self, _func: Callable = None) -> Union[int, float, complex]:
r"""Returns average value of list
Parameter
@ -99,7 +65,7 @@ class IterableABC(ABC, list):
pass
@abstractmethod
def count(self, func: Callable = None) -> int:
def count(self, _func: Callable = None) -> int:
r"""Returns length of list or count of found elements
Parameter
@ -114,7 +80,7 @@ class IterableABC(ABC, list):
pass
@abstractmethod
def distinct(self, func: Callable = None) -> 'IterableABC':
def distinct(self, _func: Callable = None) -> 'QueryableABC':
r"""Returns list without redundancies
Parameter
@ -124,7 +90,7 @@ class IterableABC(ABC, list):
Returns
-------
:class: `cpl_query.extension.iterable_abc.IterableABC`
:class: `cpl_query.base.queryable_abc.QueryableABC`
"""
pass
@ -158,18 +124,36 @@ class IterableABC(ABC, list):
"""
pass
def extend(self, __iterable: Iterable) -> 'IterableABC':
r"""Adds elements of given list to list
@abstractmethod
def first(self) -> any:
r"""Returns first element
Returns
-------
First element of list: any
"""
pass
@abstractmethod
def first_or_default(self) -> any:
r"""Returns first element or None
Returns
-------
First element of list: Optional[any]
"""
pass
@abstractmethod
def for_each(self, _func: Callable = None):
r"""Runs given function for each element of list
Parameter
---------
__iterable: :class: `cpl_query.extension.iterable.Iterable`
index
func: :class: `Callable`
function to call
"""
for value in __iterable:
self.append(value)
return self
pass
@abstractmethod
def last(self) -> any:
@ -192,39 +176,8 @@ class IterableABC(ABC, list):
pass
@abstractmethod
def first(self) -> any:
r"""Returns first element
Returns
-------
First element of list: any
"""
pass
@abstractmethod
def first_or_default(self) -> any:
r"""Returns first element or None
Returns
-------
First element of list: Optional[any]
"""
pass
@abstractmethod
def for_each(self, func: Callable):
r"""Runs given function for each element of list
Parameter
---------
func: :class: `Callable`
function to call
"""
pass
@abstractmethod
def max(self, func: Callable = None) -> Union[int, float, complex]:
r"""Returns highest value
def max(self, _func: Callable = None) -> Union[int, float, complex]:
r"""Returns the highest value
Parameter
---------
@ -238,8 +191,18 @@ class IterableABC(ABC, list):
pass
@abstractmethod
def min(self, func: Callable = None) -> Union[int, float, complex]:
r"""Returns highest value
def median(self) -> Union[int, float]:
r"""Return the median value of data elements
Returns
-------
Union[int, float]
"""
pass
@abstractmethod
def min(self, _func: Callable = None) -> Union[int, float, complex]:
r"""Returns the lowest value
Parameter
---------
@ -253,7 +216,7 @@ class IterableABC(ABC, list):
pass
@abstractmethod
def order_by(self, func: Callable) -> 'IterableABC':
def order_by(self, _func: Callable = None) -> 'QueryableABC':
r"""Sorts elements by function in ascending order
Parameter
@ -263,12 +226,12 @@ class IterableABC(ABC, list):
Returns
-------
:class: `cpl_query.extension.iterable_abc.IterableABC`
:class: `cpl_query.base.queryable_abc.QueryableABC`
"""
pass
@abstractmethod
def order_by_descending(self, func: Callable) -> 'IterableABC':
def order_by_descending(self, _func: Callable = None) -> 'QueryableABC':
r"""Sorts elements by function in descending order
Parameter
@ -278,35 +241,37 @@ class IterableABC(ABC, list):
Returns
-------
:class: `cpl_query.extension.iterable_abc.IterableABC`
:class: `cpl_query.base.queryable_abc.QueryableABC`
"""
pass
@abstractmethod
def reverse(self) -> 'IterableABC':
def reverse(self) -> 'QueryableABC':
r"""Reverses list
Returns
-------
:class: `cpl_query.extension.iterable_abc.IterableABC`
:class: `cpl_query.base.queryable_abc.QueryableABC`
"""
pass
def select(self, _f: Callable) -> 'IterableABC':
@abstractmethod
def select(self, _f: Callable) -> 'QueryableABC':
r"""Formats each element of list to a given format
Returns
-------
:class: `cpl_query.extension.iterable_abc.IterableABC`
:class: `cpl_query.base.queryable_abc.QueryableABC`
"""
pass
def select_many(self, _f: Callable) -> 'IterableABC':
@abstractmethod
def select_many(self, _f: Callable) -> 'QueryableABC':
r"""Flattens resulting lists to one
Returns
-------
:class: `cpl_query.extension.iterable_abc.IterableABC`
:class: `cpl_query.base.queryable_abc.QueryableABC`
"""
pass
@ -336,7 +301,7 @@ class IterableABC(ABC, list):
pass
@abstractmethod
def skip(self, index: int) -> 'IterableABC':
def skip(self, index: int) -> 'QueryableABC':
r"""Skips all elements from index
Parameter
@ -346,12 +311,12 @@ class IterableABC(ABC, list):
Returns
-------
:class: `cpl_query.extension.iterable_abc.IterableABC`
:class: `cpl_query.base.queryable_abc.QueryableABC`
"""
pass
@abstractmethod
def skip_last(self, index: int) -> 'IterableABC':
def skip_last(self, index: int) -> 'QueryableABC':
r"""Skips all elements after index
Parameter
@ -361,12 +326,12 @@ class IterableABC(ABC, list):
Returns
-------
:class: `cpl_query.extension.iterable_abc.IterableABC`
:class: `cpl_query.base.queryable_abc.QueryableABC`
"""
pass
@abstractmethod
def sum(self, func: Callable = None) -> Union[int, float, complex]:
def sum(self, _func: Callable = None) -> Union[int, float, complex]:
r"""Sum of all values
Parameter
@ -381,7 +346,7 @@ class IterableABC(ABC, list):
pass
@abstractmethod
def take(self, index: int) -> 'IterableABC':
def take(self, index: int) -> 'QueryableABC':
r"""Takes all elements from index
Parameter
@ -391,12 +356,12 @@ class IterableABC(ABC, list):
Returns
-------
:class: `cpl_query.extension.iterable_abc.IterableABC`
:class: `cpl_query.base.queryable_abc.QueryableABC`
"""
pass
@abstractmethod
def take_last(self, index: int) -> 'IterableABC':
def take_last(self, index: int) -> 'QueryableABC':
r"""Takes all elements after index
Parameter
@ -406,21 +371,12 @@ class IterableABC(ABC, list):
Returns
-------
:class: `cpl_query.extension.iterable_abc.IterableABC`
:class: `cpl_query.base.queryable_abc.QueryableABC`
"""
pass
def to_list(self) -> list:
r"""Converts :class: `cpl_query.extension.iterable_abc.IterableABC` to :class: `list`
Returns
-------
:class: `list`
"""
return list(self)
@abstractmethod
def where(self, func: Callable) -> 'IterableABC':
def where(self, _func: Callable = None) -> 'QueryableABC':
r"""Select element by function
Parameter
@ -430,6 +386,6 @@ class IterableABC(ABC, list):
Returns
-------
:class: `cpl_query.extension.iterable_abc.IterableABC`
:class: `cpl_query.base.queryable_abc.QueryableABC`
"""
pass

49
src/cpl_query/base/sequence_values.py Normal file
View File

@ -0,0 +1,49 @@
import io
import itertools
from cpl_query.exceptions import IndexOutOfRangeException
class SequenceValues:
def __init__(self, data, _t: type):
if data is None:
data = []
if len(data) > 0:
def type_check(_t: type, _l: list):
return all(isinstance(x, _t) for x in _l)
if not type_check(_t, data):
raise Exception(f'Unexpected type\nExpected type: {_t}')
if not hasattr(data, '__iter__'):
raise TypeError(f'{type(self).__name__} must be instantiated with an iterable object')
self._new_cycle = lambda: itertools.cycle(data)
self._len = lambda: len(data)
self._index = 0
self._cycle = self._new_cycle()
def __len__(self):
return self._len()
def __iter__(self):
i = 0
while i < len(self):
yield next(self._cycle)
i += 1
def __next__(self):
if self._index >= len(self):
raise IndexOutOfRangeException()
self._index += 1
return self.next()
def next(self):
return next(self._cycle)
def reset(self):
self._index = 0
self._cycle = self._new_cycle()

2
src/cpl_query/cpl-query.json
View File

@ -4,7 +4,7 @@
"Version": {
"Major": "2022",
"Minor": "10",
"Micro": "2"
"Micro": "9"
},
"Author": "Sven Heidemann",
"AuthorEmail": "sven.heidemann@sh-edraft.de",

30
src/cpl_query/enumerable/__init__.py Normal file
View File

@ -0,0 +1,30 @@
# -*- coding: utf-8 -*-
"""
cpl-query sh-edraft Common Python library Query
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library Python integrated Queries
:copyright: (c) 2021 - 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_query.enumerable'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2021 - 2022 sh-edraft.de'
__version__ = '2022.10.9'
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='9')

319
src/cpl_query/enumerable/enumerable.py Normal file
View File

@ -0,0 +1,319 @@
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):
return x
class Enumerable(EnumerableABC):
r"""Implementation of :class: `cpl_query.enumerable.enumerable_abc.EnumerableABC`
"""
def __init__(self, t: type = None, values: Union[list, iter] = 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()
if _index >= len(_list):
raise IndexOutOfRangeException()
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
if index >= len(self) or index < 0:
raise IndexOutOfRangeException()
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()
if _index >= len(_list):
raise IndexOutOfRangeException()
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
if index >= len(_list) or index < 0:
raise IndexOutOfRangeException()
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)))

106
src/cpl_query/enumerable/enumerable_abc.py Normal file
View File

@ -0,0 +1,106 @@
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):
r"""ABC to define functions on list
"""
@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])
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
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`
Returns
-------
:class: `cpl_query.iterable.iterable_abc.IterableABC`
"""
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 Normal file
View File

@ -0,0 +1,36 @@
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 Normal file
View File

@ -0,0 +1,43 @@
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

4
src/cpl_query/exceptions.py
View File

@ -23,8 +23,8 @@ class IndexOutOfRangeException(Exception):
r"""Exception when index is out of range
"""
def __init__(self):
Exception.__init__(self, f'List index out of range')
def __init__(self, err: str = None):
Exception.__init__(self, f'List index out of range' if err is None else err)
class InvalidTypeException(Exception):

8
src/cpl_query/extension/__init__.py
View File

@ -15,17 +15,13 @@ __title__ = 'cpl_query.extension'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2021 - 2022 sh-edraft.de'
__version__ = '2022.10.2'
__version__ = '2022.10.9'
from collections import namedtuple
# imports:
from .iterable_abc import IterableABC
from .iterable import Iterable
from .list import List
from .ordered_iterable_abc import OrderedIterableABC
from .ordered_iterable import OrderedIterable
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='10', micro='2')
version_info = VersionInfo(major='2022', minor='10', micro='9')

106
src/cpl_query/extension/iterable.py
View File

@ -1,106 +0,0 @@
from typing import Callable, Optional, Union
from cpl_query.extension.iterable_abc import IterableABC
from cpl_query.extension.ordered_iterable_abc import OrderedIterableABC
from cpl_query.query import Query
class Iterable(IterableABC):
def __init__(self, t: type = None, values: list = None):
IterableABC.__init__(self, t, values)
def any(self, func: Callable) -> bool:
return Query.any(self, func)
def all(self, func: Callable) -> bool:
return Query.all(self, func)
def average(self, func: Callable = None) -> Union[int, float, complex]:
return Query.avg(self, func)
def contains(self, value: object) -> bool:
return Query.contains(self, value)
def count(self, func: Callable = None) -> int:
return Query.count(self, func)
def distinct(self, func: Callable = None) -> IterableABC:
return self.__to_self(Query.distinct(self, func))
def element_at(self, index: int) -> any:
return Query.element_at(self, index)
def element_at_or_default(self, index: int) -> Optional[any]:
return Query.element_at_or_default(self, index)
def last(self) -> any:
return Query.last(self)
def last_or_default(self) -> Optional[any]:
return Query.last_or_default(self)
def first(self) -> any:
return Query.first(self)
def first_or_default(self) -> Optional[any]:
return Query.first_or_default(self)
def for_each(self, func: Callable):
Query.for_each(self, func)
def max(self, func: Callable = None) -> Union[int, float, complex]:
return Query.max(self, func)
def min(self, func: Callable = None) -> Union[int, float, complex]:
return Query.min(self, func)
def order_by(self, func: Callable) -> OrderedIterableABC:
res = Query.order_by(self, func)
from cpl_query.extension.ordered_iterable import OrderedIterable
res.__class__ = OrderedIterable
return res
def order_by_descending(self, func: Callable) -> OrderedIterableABC:
res = Query.order_by_descending(self, func)
from cpl_query.extension.ordered_iterable import OrderedIterable
res.__class__ = OrderedIterable
return res
def reverse(self) -> IterableABC:
return Query.reverse(self)
def single(self) -> any:
return Query.single(self)
def single_or_default(self) -> Optional[any]:
return Query.single_or_default(self)
def select(self, _f: Callable) -> IterableABC:
return self.__to_self(Query.select(self, _f))
def select_many(self, _f: Callable) -> IterableABC:
return self.__to_self(Query.select_many(self, _f))
def skip(self, index: int) -> IterableABC:
return self.__to_self(Query.skip(self, index))
def skip_last(self, index: int) -> IterableABC:
return self.__to_self(Query.skip_last(self, index))
def sum(self, func: Callable = None) -> Union[int, float, complex]:
return Query.sum(self, func)
def take(self, index: int) -> IterableABC:
return self.__to_self(Query.take(self, index))
def take_last(self, index: int) -> IterableABC:
return self.__to_self(Query.take_last(self, index))
def where(self, func: Callable) -> IterableABC:
return self.__to_self(Query.where(self, func))
@staticmethod
def __to_self(obj: IterableABC) -> IterableABC:
obj.__class__ = Iterable
return obj

26
src/cpl_query/extension/list.py
View File

@ -1,9 +1,31 @@
from cpl_query.extension.iterable import Iterable
from typing import Iterable as IterableType
from cpl_query.iterable.iterable import Iterable
class List(Iterable):
r"""Implementation of :class: `cpl_query.extension.iterable.Iterable`
"""
def __init__(self, t: type = None, values: list = None):
def __init__(self, t: type = None, values: IterableType = None):
Iterable.__init__(self, t, values)
def to_enumerable(self) -> 'EnumerableABC':
r"""Converts :class: `cpl_query.iterable.iterable_abc.IterableABC` to :class: `cpl_query.enumerable.enumerable_abc.EnumerableABC`
Returns
-------
:class: `cpl_query.enumerable.enumerable_abc.EnumerableABC`
"""
from cpl_query.enumerable.enumerable import Enumerable
return Enumerable(self._type, self.to_list())
def to_iterable(self) -> 'IterableABC':
r"""Converts :class: `cpl_query.enumerable.enumerable_abc.EnumerableABC` to :class: `cpl_query.iterable.iterable_abc.IterableABC`
Returns
-------
:class: `cpl_query.iterable.iterable_abc.IterableABC`
"""
from cpl_query.iterable.iterable import Iterable
return Iterable(self._type, self.to_list())

22
src/cpl_query/extension/ordered_iterable.py
View File

@ -1,22 +0,0 @@
from collections.abc import Callable
from cpl_query.extension.iterable import Iterable
from cpl_query.extension.ordered_iterable_abc import OrderedIterableABC
from cpl_query.query import Query
class OrderedIterable(Iterable, OrderedIterableABC):
r"""Implementation of :class: `cpl_query.extension.Iterable` `cpl_query.extension.OrderedIterableABC`
"""
def __init__(self, _t: type = None):
Iterable.__init__(self, _t)
OrderedIterableABC.__init__(self, _t)
def then_by(self, _func: Callable) -> OrderedIterableABC:
self._funcs.append(_func)
return Query.then_by(self, lambda *args: [f(*args) for f in self._funcs])
def then_by_descending(self, _func: Callable) -> OrderedIterableABC:
self._funcs.append(_func)
return Query.then_by_descending(self, lambda *args: [f(*args) for f in self._funcs])

30
src/cpl_query/iterable/__init__.py Normal file
View File

@ -0,0 +1,30 @@
# -*- coding: utf-8 -*-
"""
cpl-query sh-edraft Common Python library Query
~~~~~~~~~~~~~~~~~~~
sh-edraft Common Python library Python integrated Queries
:copyright: (c) 2021 - 2022 sh-edraft.de
:license: MIT, see LICENSE for more details.
"""
__title__ = 'cpl_query.iterable'
__author__ = 'Sven Heidemann'
__license__ = 'MIT'
__copyright__ = 'Copyright (c) 2021 - 2022 sh-edraft.de'
__version__ = '2022.10.9'
from collections import namedtuple
# imports:
from .iterable_abc import IterableABC
from .iterable import Iterable
from .ordered_iterable_abc import OrderedIterableABC
from .ordered_iterable import OrderedIterable
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
version_info = VersionInfo(major='2022', minor='10', micro='9')

343
src/cpl_query/iterable/iterable.py Normal file
View File

@ -0,0 +1,343 @@
from typing import Callable, Optional, Union, Iterable as IterableType
from cpl_query._helper import is_number
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument, InvalidTypeException, IndexOutOfRangeException
from cpl_query.iterable.iterable_abc import IterableABC
from cpl_query.iterable.ordered_iterable_abc import OrderedIterableABC
def _default_lambda(x: object):
return x
class Iterable(IterableABC):
def __init__(self, t: type = None, values: IterableType = None):
IterableABC.__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) -> IterableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
_func = _default_lambda
result = Iterable()
known_values = []
for element in self:
value = _func(element)
if value in known_values:
continue
known_values.append(value)
result.append(element)
return result
def element_at(self, _index: int) -> any:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
return self[_index]
def element_at_or_default(self, _index: int) -> any:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
try:
return self[_index]
except IndexError:
return None
def first(self: IterableABC) -> any:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if len(self) == 0:
raise IndexOutOfRangeException()
return self[0]
def first_or_default(self: IterableABC) -> Optional[any]:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if len(self) == 0:
return None
return self[0]
def last(self: IterableABC) -> any:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if len(self) == 0:
raise IndexOutOfRangeException()
return self[len(self) - 1]
def last_or_default(self: IterableABC) -> Optional[any]:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if len(self) == 0:
return None
return self[len(self) - 1]
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 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) -> OrderedIterableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
_func = _default_lambda
from cpl_query.iterable.ordered_iterable import OrderedIterable
return OrderedIterable(self.type, _func, sorted(self, key=_func))
def order_by_descending(self, _func: Callable = None) -> OrderedIterableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
_func = _default_lambda
from cpl_query.iterable.ordered_iterable import OrderedIterable
return OrderedIterable(self.type, _func, sorted(self, key=_func, reverse=True))
def reverse(self: IterableABC) -> IterableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
return Iterable().extend(reversed(self.to_list()))
def select(self, _func: Callable = None) -> IterableABC:
if _func is None:
_func = _default_lambda
result = Iterable()
result.extend(_func(_o) for _o in self)
return result
def select_many(self, _func: Callable = None) -> IterableABC:
if _func is None:
_func = _default_lambda
result = Iterable()
# The line below is pain. I don't understand anything of it...
# written on 09.11.2022 by Sven Heidemann
elements = [_a for _o in self for _a in _func(_o)]
result.extend(elements)
return result
def single(self: IterableABC) -> any:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if len(self) > 1:
raise Exception('Found more than one element')
elif len(self) == 0:
raise Exception('Found no element')
return self[0]
def single_or_default(self: IterableABC) -> Optional[any]:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if len(self) > 1:
raise Exception('Index out of range')
elif len(self) == 0:
return None
return self[0]
def skip(self, _index: int) -> IterableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
if _index >= len(self):
raise IndexOutOfRangeException()
return Iterable(self.type, values=self[_index:])
def skip_last(self, _index: int) -> IterableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
index = len(self) - _index
if index >= len(self) or index < 0:
raise IndexOutOfRangeException()
result = Iterable()
result.extend(self[:index])
return result
def take(self, _index: int) -> IterableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
if _index >= len(self):
raise IndexOutOfRangeException()
result = Iterable()
result.extend(self[:_index])
return result
def take_last(self, _index: int) -> IterableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
index = len(self) - _index
if index >= len(self) or index < 0:
raise IndexOutOfRangeException()
result = Iterable()
result.extend(self[index:])
return result
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) -> IterableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
if _func is None:
_func = _default_lambda
result = Iterable(self.type)
for element in self:
if _func(element):
result.append(element)
return result

71
src/cpl_query/iterable/iterable_abc.py Normal file
View File

@ -0,0 +1,71 @@
from abc import abstractmethod
from typing import Iterable
from cpl_query.base.queryable_abc import QueryableABC
class IterableABC(list, QueryableABC):
r"""ABC to define functions on list
"""
@abstractmethod
def __init__(self, t: type = None, values: Iterable = None):
self._type = t
list.__init__(self, [] if values is None else values)
def __repr__(self):
return f'<{type(self).__name__} {list(self).__repr__()}>'
@property
def type(self) -> type:
return self._type
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]
def __str__(self):
return str(self.to_list())
def append(self, __object: object) -> None:
r"""Adds element to list
Parameter
---------
__object: :class:`object`
value
"""
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)
super().append(__object)
def extend(self, __iterable: Iterable) -> 'IterableABC':
r"""Adds elements of given list to list
Parameter
---------
__iterable: :class: `cpl_query.extension.iterable.Iterable`
index
"""
for value in __iterable:
self.append(value)
return self
def to_enumerable(self) -> 'EnumerableABC':
r"""Converts :class: `cpl_query.iterable.iterable_abc.IterableABC` to :class: `cpl_query.enumerable.enumerable_abc.EnumerableABC`
Returns
-------
:class: `cpl_query.enumerable.enumerable_abc.EnumerableABC`
"""
from cpl_query.enumerable.enumerable import Enumerable
return Enumerable(self._type, self.to_list())

35
src/cpl_query/iterable/ordered_iterable.py Normal file
View File

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

11
src/cpl_query/extension/ordered_iterable_abc.py → src/cpl_query/iterable/ordered_iterable_abc.py
View File

@ -1,14 +1,15 @@
from abc import abstractmethod
from collections.abc import Callable
from typing import Iterable
from cpl_query.extension.iterable_abc import IterableABC
from cpl_query.iterable.iterable_abc import IterableABC
class OrderedIterableABC(IterableABC):
@abstractmethod
def __init__(self, _t: type, _func: Callable = None):
IterableABC.__init__(self, _t)
def __init__(self, _t: type, _func: Callable = None, _values: Iterable = None):
IterableABC.__init__(self, _t, _values)
self._funcs: list[Callable] = []
if _func is not None:
self._funcs.append(_func)
@ -23,7 +24,7 @@ class OrderedIterableABC(IterableABC):
Returns
-------
list of :class:`cpl_query.extension.OrderedIterableABC`
list of :class:`cpl_query.iterable.ordered_iterable_abc.OrderedIterableABC`
"""
pass
@ -37,6 +38,6 @@ class OrderedIterableABC(IterableABC):
Returns
-------
list of :class:`cpl_query.extension.OrderedIterableABC`
list of :class:`cpl_query.iterable.ordered_iterable_abc.OrderedIterableABC`
"""
pass

405
src/cpl_query/query.py
View File

@ -1,405 +0,0 @@
from typing import Callable, Union, Optional
from cpl_query._helper import is_number
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument, InvalidTypeException, IndexOutOfRangeException
from cpl_query.extension.iterable_abc import IterableABC
from cpl_query.extension.ordered_iterable_abc import OrderedIterableABC
class Query:
@staticmethod
def all(_list: IterableABC, _func: Callable) -> bool:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
result = Query.where(_list, _func)
return len(result) == len(_list)
@staticmethod
def any(_list: IterableABC, _func: Callable) -> bool:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
result = Query.where(_list, _func)
return len(result) > 0
@staticmethod
def avg(_list: IterableABC, _func: Callable) -> Union[int, float, complex]:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None and not is_number(_list.type):
raise InvalidTypeException()
average = 0
count = len(_list)
for element in _list:
if _func is not None:
value = _func(element)
else:
value = element
average += value
return average / count
@staticmethod
def contains(_list: IterableABC, _value: object) -> bool:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _value is None:
raise ArgumentNoneException(ExceptionArgument.value)
return _value in _list
@staticmethod
def count(_list: IterableABC, _func: Callable = None) -> int:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
return len(_list)
return len(Query.where(_list, _func))
@staticmethod
def distinct(_list: IterableABC, _func: Callable) -> IterableABC:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
result = IterableABC()
known_values = []
for element in _list:
value = _func(element)
if value in known_values:
continue
known_values.append(value)
result.append(element)
return result
@staticmethod
def element_at(_list: IterableABC, _index: int) -> any:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
return _list[_index]
@staticmethod
def element_at_or_default(_list: IterableABC, _index: int) -> any:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
try:
return _list[_index]
except IndexError:
return None
@staticmethod
def first(_list: IterableABC) -> any:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if len(_list) == 0:
raise IndexOutOfRangeException()
return _list[0]
@staticmethod
def first_or_default(_list: IterableABC) -> Optional[any]:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if len(_list) == 0:
return None
return _list[0]
@staticmethod
def last(_list: IterableABC) -> any:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if len(_list) == 0:
raise IndexOutOfRangeException()
return _list[len(_list) - 1]
@staticmethod
def last_or_default(_list: IterableABC) -> Optional[any]:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if len(_list) == 0:
return None
return _list[len(_list) - 1]
@staticmethod
def for_each(_list: IterableABC, _func: Callable):
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
for element in _list:
_func(element)
@staticmethod
def max(_list: IterableABC, _func: Callable) -> Union[int, float, complex]:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None and not is_number(_list.type):
raise InvalidTypeException()
max_value = 0
for element in _list:
if _func is not None:
value = _func(element)
else:
value = element
if value > max_value:
max_value = value
return max_value
@staticmethod
def min(_list: IterableABC, _func: Callable) -> Union[int, float, complex]:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None and not is_number(_list.type):
raise InvalidTypeException()
min_value = 0
is_first = True
for element in _list:
if _func is not None:
value = _func(element)
else:
value = element
if is_first:
min_value = value
is_first = False
if value < min_value:
min_value = value
return min_value
@staticmethod
def order_by(_list: IterableABC, _func: Callable) -> OrderedIterableABC:
result = OrderedIterableABC(_list.type, _func)
_list.sort(key=_func)
result.extend(_list)
return result
@staticmethod
def order_by_descending(_list: IterableABC, _func: Callable) -> OrderedIterableABC:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
result = OrderedIterableABC(_list.type, _func)
_list.sort(key=_func, reverse=True)
result.extend(_list)
return result
@staticmethod
def then_by(_list: OrderedIterableABC, _func: Callable) -> OrderedIterableABC:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
_list.sort(key=_func)
return _list
@staticmethod
def then_by_descending(_list: OrderedIterableABC, _func: Callable) -> OrderedIterableABC:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
_list.sort(key=_func, reverse=True)
return _list
@staticmethod
def reverse(_list: IterableABC) -> IterableABC:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
result = IterableABC()
_copied_list = _list.to_list()
_copied_list.reverse()
for element in _copied_list:
result.append(element)
return result
@staticmethod
def select(_list: IterableABC, _f: Callable) -> any:
result = IterableABC()
result.extend(_f(_o) for _o in _list)
return result
@staticmethod
def select_many(_list: IterableABC, _f: Callable) -> any:
result = IterableABC()
# The line below is pain. I don't understand anything of it...
# written on 09.11.2022 by Sven Heidemann
elements = [_a for _o in _list for _a in _f(_o)]
result.extend(elements)
return result
@staticmethod
def single(_list: IterableABC) -> any:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if len(_list) > 1:
raise Exception('Found more than one element')
elif len(_list) == 0:
raise Exception('Found no element')
return _list[0]
@staticmethod
def single_or_default(_list: IterableABC) -> Optional[any]:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if len(_list) > 1:
raise Exception('Index out of range')
elif len(_list) == 0:
return None
return _list[0]
@staticmethod
def skip(_list: IterableABC, _index: int) -> IterableABC:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
if _index >= len(_list):
raise IndexOutOfRangeException()
result = IterableABC()
result.extend(_list[_index:])
return result
@staticmethod
def skip_last(_list: IterableABC, _index: int) -> IterableABC:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
index = len(_list) - _index
if index >= len(_list) or index < 0:
raise IndexOutOfRangeException()
result = IterableABC()
result.extend(_list[:index])
return result
@staticmethod
def take(_list: IterableABC, _index: int) -> IterableABC:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
if _index >= len(_list):
raise IndexOutOfRangeException()
result = IterableABC()
result.extend(_list[:_index])
return result
@staticmethod
def take_last(_list: IterableABC, _index: int) -> IterableABC:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
index = len(_list) - _index
if index >= len(_list) or index < 0:
raise IndexOutOfRangeException()
result = IterableABC()
result.extend(_list[index:])
return result
@staticmethod
def sum(_list: IterableABC, _func: Callable) -> Union[int, float, complex]:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None and not is_number(_list.type):
raise InvalidTypeException()
result = 0
for element in _list:
if _func is not None:
value = _func(element)
else:
value = element
result += value
return result
@staticmethod
def where(_list: IterableABC, _func: Callable) -> IterableABC:
if _list is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
result = IterableABC(_list.type)
for element in _list:
if _func(element):
result.append(element)
return result

359
unittests/unittests_query/enumerable_query_test_case.py Normal file
View File

@ -0,0 +1,359 @@
import string
import unittest
from random import randint
from cpl_core.utils import String
from cpl_query.enumerable.enumerable import Enumerable
from cpl_query.exceptions import InvalidTypeException, ArgumentNoneException, IndexOutOfRangeException
from unittests_query.models import User, Address
class EnumerableQueryTestCase(unittest.TestCase):
def setUp(self) -> None:
self._tests = Enumerable(User)
self._t_user = User(
'Test user',
Address(
'teststr.',
15
)
)
self._t_user2 = User(
'Test user',
Address(
'teststr.',
14
)
)
self._generate_test_data()
def _generate_test_data(self):
for i in range(0, 100):
user = User(
String.random_string(string.ascii_letters, 8).lower(),
Address(
String.random_string(string.ascii_letters, 10).lower(),
randint(1, 10)
)
)
self._tests.add(user)
self._tests.add(self._t_user)
self._tests.add(self._t_user2)
def test_any(self):
results = []
for user in self._tests:
if user.address.nr == 10:
results.append(user)
res = self._tests.any(lambda u: u.address.nr == 10)
n_res = self._tests.any(lambda u: u.address.nr == 100)
self.assertTrue(res)
self.assertFalse(n_res)
def test_all(self):
results = []
for user in self._tests:
if user.address.nr == 10:
results.append(user)
res = self._tests.all(lambda u: u.address is not None)
n_res = self._tests.all(lambda u: u.address.nr == 100)
self.assertTrue(res)
self.assertFalse(n_res)
def test_avg(self):
avg = 0
for user in self._tests:
avg += user.address.nr
avg = avg / len(self._tests)
res = self._tests.average(lambda u: u.address.nr)
self.assertEqual(avg, res)
def invalid():
tests = Enumerable(str, ['hello', 'world'])
e_res = tests.average()
self.assertRaises(InvalidTypeException, invalid)
tests = Enumerable(int, list(range(0, 100)))
self.assertEqual(sum(tests) / len(tests), tests.average())
def wrong2():
tests2 = Enumerable(int, values=list(range(0, 100)))
e_res = tests2.average(lambda u: u.address.nr)
self.assertRaises(AttributeError, wrong2)
def test_contains(self):
self.assertTrue(self._tests.contains(self._t_user))
self.assertFalse(self._tests.contains(User("Test", None)))
def test_count(self):
self.assertEqual(len(self._tests), self._tests.count())
self.assertEqual(1, self._tests.count(lambda u: u == self._t_user))
def test_distinct(self):
res = self._tests.distinct(lambda u: u.address.nr).where(lambda u: u.address.nr == 5)
self.assertEqual(1, len(res))
def test_element_at(self):
index = randint(0, len(self._tests) - 1)
self.assertEqual(self._tests.element_at(index), self._tests.element_at(index))
def invalid():
result = self._tests.element_at(len(self._tests))
self.assertRaises(IndexOutOfRangeException, invalid)
def test_element_at_or_default(self):
index = randint(0, len(self._tests) - 1)
self.assertEqual(self._tests.element_at(index), self._tests.element_at_or_default(index))
self.assertIsNone(self._tests.element_at_or_default(len(self._tests)))
def test_last(self):
results = []
for user in self._tests:
if user.address.nr == 10:
results.append(user)
res = self._tests.where(lambda u: u.address.nr == 10)
s_res = self._tests.where(lambda u: u.address.nr == 10).last()
self.assertEqual(len(res), len(results))
self.assertEqual(res.element_at(len(res) - 1), s_res)
def test_last_or_default(self):
results = []
for user in self._tests:
if user.address.nr == 10:
results.append(user)
res = self._tests.where(lambda u: u.address.nr == 10)
s_res = self._tests.where(lambda u: u.address.nr == 10).last_or_default()
sn_res = self._tests.where(lambda u: u.address.nr == 11).last_or_default()
self.assertEqual(len(res), len(results))
self.assertEqual(res.element_at(len(res) - 1), s_res)
self.assertIsNone(sn_res)
def test_first(self):
results = []
for user in self._tests:
if user.address.nr == 10:
results.append(user)
res = self._tests.where(lambda u: u.address.nr == 10)
s_res = self._tests.where(lambda u: u.address.nr == 10).first()
self.assertEqual(len(res), len(results))
self.assertEqual(res.element_at(0), s_res)
self.assertEqual(res.element_at(0), res.first())
self.assertEqual(res.first(), res.first())
def test_first_or_default(self):
results = []
for user in self._tests:
if user.address.nr == 10:
results.append(user)
res = self._tests.where(lambda u: u.address.nr == 10)
s_res = self._tests.where(lambda u: u.address.nr == 10).first_or_default()
sn_res = self._tests.where(lambda u: u.address.nr == 11).first_or_default()
self.assertEqual(len(res), len(results))
self.assertEqual(res.element_at(0), s_res)
self.assertIsNone(sn_res)
def test_for_each(self):
users = []
self._tests.for_each(lambda user: (
users.append(user)
)
)
self.assertEqual(len(users), len(self._tests))
def test_max(self):
res = self._tests.max(lambda u: u.address.nr)
self.assertEqual(res, self._t_user.address.nr)
tests = Enumerable(values=list(range(0, 100)))
self.assertEqual(99, tests.max())
def invalid():
tests = Enumerable(str, ['hello', 'world'])
e_res = tests.average()
self.assertRaises(InvalidTypeException, invalid)
def test_min(self):
res = self._tests.min(lambda u: u.address.nr)
self.assertEqual(1, res)
tests = Enumerable(values=list(range(0, 100)))
self.assertEqual(0, tests.min())
def invalid():
tests = Enumerable(str, ['hello', 'world'])
e_res = tests.average()
self.assertRaises(InvalidTypeException, invalid)
def test_order_by(self):
res = self._tests.order_by(lambda user: user.address.street)
res2 = self._tests.order_by(lambda user: user.address.nr).to_list()
s_res = self._tests.to_list()
s_res.sort(key=lambda user: user.address.street)
self.assertEqual(res.to_list(), s_res)
s_res = self._tests.to_list()
s_res.sort(key=lambda user: user.address.nr)
self.assertEqual(res2, s_res)
self.assertEqual(self._t_user, res.where(lambda u: u.address.nr == self._t_user.address.nr).single())
def test_order_by_descending(self):
res = self._tests.order_by_descending(lambda user: user.address.street).to_list()
res2 = self._tests.order_by_descending(lambda user: user.address.nr).to_list()
s_res = self._tests.to_list()
s_res.sort(key=lambda user: user.address.street, reverse=True)
self.assertEqual(res, s_res)
s_res = self._tests.to_list()
s_res.sort(key=lambda user: user.address.nr, reverse=True)
self.assertEqual(res2, s_res)
def test_then_by(self):
res = self._tests.order_by(lambda user: user.address.street).then_by(lambda user: user.address.nr).to_list()
s_res = self._tests.to_list()
s_res.sort(key=lambda user: (user.address.street, user.address.nr))
self.assertEqual(res, s_res)
def test_then_by_descending(self):
res = self._tests.order_by_descending(lambda user: user.address.street).then_by_descending(lambda user: user.address.nr).to_list()
s_res = self._tests.to_list()
s_res.sort(key=lambda user: (user.address.street, user.address.nr), reverse=True)
self.assertEqual(res, s_res)
def test_reverse(self):
res = self._tests.reverse().to_list()
l_res = self._tests.to_list()
l_res.reverse()
self.assertEqual(res, l_res)
def test_select(self):
range_list = Enumerable(int, range(0, 100))
selected_range = range_list.select(lambda x: x + 1)
modulo_range = []
for x in range(0, 100):
if x % 2 == 0:
modulo_range.append(x)
self.assertEqual(selected_range.to_list(), list(range(1, 101)))
self.assertEqual(range_list.where(lambda x: x % 2 == 0).to_list(), modulo_range)
def test_select_many(self):
range_list = Enumerable(int, list(range(0, 100)))
selected_range = range_list.select(lambda x: [x, x])
self.assertEqual(selected_range.to_list(), [[x, x] for x in range(0, 100)])
self.assertEqual(selected_range.select_many(lambda x: x).to_list(), [_x for _l in [2 * [x] for x in range(0, 100)] for _x in _l])
class TestClass:
def __init__(self, i, is_sub=False):
self.i = i
if is_sub:
return
self.elements = [TestClass(x, True) for x in range(0, 10)]
elements = Enumerable(TestClass, [TestClass(i) for i in range(0, 100)])
selected_elements = elements.select_many(lambda x: x.elements).select(lambda x: x.i)
self.assertEqual(selected_elements.where(lambda x: x == 0).count(), 100)
def test_single(self):
res = self._tests.where(lambda u: u.address.nr == self._t_user.address.nr)
s_res = self._tests.where(lambda u: u.address.nr == self._t_user.address.nr).single()
self.assertEqual(len(res), 1)
self.assertEqual(self._t_user, s_res)
def test_single_or_default(self):
res = self._tests.where(lambda u: u.address.nr == self._t_user.address.nr)
s_res = self._tests.where(lambda u: u.address.nr == self._t_user.address.nr).single_or_default()
sn_res = self._tests.where(lambda u: u.address.nr == self._t_user.address.nr + 1).single_or_default()
self.assertEqual(len(res), 1)
self.assertEqual(self._t_user, s_res)
self.assertIsNone(sn_res)
def test_skip(self):
skipped = self._tests.skip(5).to_list()
self.assertEqual(len(skipped), len(self._tests) - 5)
self.assertEqual(skipped, self._tests.to_list()[5:])
def test_skip_last(self):
skipped = self._tests.skip_last(5)
self.assertEqual(skipped.count(), len(self._tests) - 5)
self.assertEqual(skipped.to_list(), self._tests.to_list()[:-5])
self.assertEqual(skipped.last(), self._tests.to_list()[:-5][len(self._tests.to_list()[:-5]) - 1])
def test_sum(self):
res = self._tests.sum(lambda u: u.address.nr)
s_res = 0
for user in self._tests:
s_res += user.address.nr
self.assertEqual(s_res, res)
tests = Enumerable(values=list(range(0, 100)))
self.assertEqual(0, tests.min())
def invalid():
tests2 = Enumerable(str, ['hello', 'world'])
e_res = tests2.average()
self.assertRaises(InvalidTypeException, invalid)
def test_take(self):
skipped = self._tests.take(5)
self.assertEqual(skipped.count(), 5)
self.assertEqual(skipped.to_list(), self._tests.to_list()[:5])
def test_take_last(self):
skipped = self._tests.take_last(5)
self.assertEqual(skipped.count(), 5)
self.assertEqual(skipped.to_list(), self._tests.to_list()[-5:])
self.assertEqual(skipped.last(), self._tests.to_list()[len(self._tests) - 1])
def test_where(self):
results = []
for user in self._tests:
if user.address.nr == 5:
results.append(user)
res = self._tests.where(lambda u: u.address.nr == 5)
self.assertEqual(len(results), len(res))
def ex():
e_res = self._tests.where(None)
self.assertRaises(ArgumentNoneException, ex)

67
unittests/unittests_query/enumerable_test_case.py Normal file
View File

@ -0,0 +1,67 @@
import unittest
from cpl_query.enumerable.enumerable import Enumerable
class EnumerableTestCase(unittest.TestCase):
def setUp(self) -> None:
self._list = Enumerable(int)
def _clear(self):
self._list.clear()
self.assertEqual(self._list, [])
def test_append(self):
self._list.add(1)
self._list.add(2)
self._list.add(3)
self.assertEqual(self._list.to_list(), [1, 2, 3])
self.assertRaises(Exception, lambda v: self._list.add(v), '3')
def test_default(self):
self.assertEqual(Enumerable.empty().to_list(), [])
self.assertEqual(Enumerable.range(0, 100).to_list(), list(range(0, 100)))
def test_iter(self):
n = 0
elements = Enumerable.range(0, 100)
while n < 100:
self.assertEqual(elements.next(), n)
n += 1
def test_for(self):
n = 0
for i in Enumerable.range(0, 100):
self.assertEqual(i, n)
n += 1
def test_get(self):
self._list.add(1)
self._list.add(2)
self._list.add(3)
self.assertEqual(self._list.element_at(2), [1, 2, 3][2])
def test_count(self):
self._list.add(1)
self._list.add(2)
self._list.add(3)
self.assertEqual(self._list.count(), 3)
def test_remove(self):
old_values = self._list._values
self._list.add(1)
self.assertNotEqual(old_values, self._list._values)
self._list.add(2)
self._list.add(3)
self.assertEqual(self._list.to_list(), [1, 2, 3])
self.assertRaises(Exception, lambda v: self._list.add(v), '3')
old_values = self._list._values
self._list.remove(3)
self.assertNotEqual(old_values, self._list._values)
self.assertEqual(self._list.to_list(), [1, 2])
self.assertRaises(Exception, lambda v: self._list.add(v), '3')

69
unittests/unittests_query/query_test_case.py → unittests/unittests_query/iterable_query_test_case.py
View File

@ -8,7 +8,7 @@ from cpl_query.extension.list import List
from unittests_query.models import User, Address
class QueryTestCase(unittest.TestCase):
class IterableQueryTestCase(unittest.TestCase):
def setUp(self) -> None:
self._tests = List(User)
@ -19,6 +19,13 @@ class QueryTestCase(unittest.TestCase):
15
)
)
self._t_user2 = User(
'Test user',
Address(
'teststr.',
14
)
)
self._generate_test_data()
@ -35,6 +42,7 @@ class QueryTestCase(unittest.TestCase):
self._tests.append(user)
self._tests.append(self._t_user)
self._tests.append(self._t_user2)
def test_any(self):
results = []
@ -143,6 +151,8 @@ class QueryTestCase(unittest.TestCase):
self.assertEqual(len(res), len(results))
self.assertEqual(res[0], s_res)
self.assertEqual(res[0], res.first())
self.assertEqual(res.first(), res.first())
def test_first_or_default(self):
results = []
@ -169,7 +179,7 @@ class QueryTestCase(unittest.TestCase):
def test_max(self):
res = self._tests.max(lambda u: u.address.nr)
self.assertEqual(self._t_user.address.nr, res)
self.assertEqual(res, self._t_user.address.nr)
tests = List(values=list(range(0, 100)))
self.assertEqual(99, tests.max())
@ -195,40 +205,41 @@ class QueryTestCase(unittest.TestCase):
def test_order_by(self):
res = self._tests.order_by(lambda user: user.address.street)
res2 = self._tests.order_by(lambda user: user.address.nr)
s_res = self._tests
res2 = self._tests.order_by(lambda user: user.address.nr).to_list()
s_res = self._tests.to_list()
s_res.sort(key=lambda user: user.address.street)
self.assertEqual(res.to_list(), s_res)
self.assertEqual(res, s_res)
s_res = self._tests.to_list()
s_res.sort(key=lambda user: user.address.nr)
self.assertEqual(res2, s_res)
self.assertEqual(self._t_user, res.where(lambda u: u.address.nr == self._t_user.address.nr).single())
def test_order_by_descending(self):
res = self._tests.order_by_descending(lambda user: user.address.street)
res2 = self._tests.order_by_descending(lambda user: user.address.nr)
s_res = self._tests
res = self._tests.order_by_descending(lambda user: user.address.street).to_list()
res2 = self._tests.order_by_descending(lambda user: user.address.nr).to_list()
s_res = self._tests.to_list()
s_res.sort(key=lambda user: user.address.street, reverse=True)
self.assertEqual(res, s_res)
s_res = self._tests.to_list()
s_res.sort(key=lambda user: user.address.nr, reverse=True)
self.assertEqual(res2, s_res)
def test_then_by(self):
res = self._tests.order_by(lambda user: user.address.street[0]).then_by(lambda user: user.address.nr)
res = self._tests.order_by(lambda user: user.address.street).then_by(lambda user: user.address.nr).to_list()
s_res = self._tests
s_res.sort(key=lambda user: (user.address.street[0], user.address.nr))
s_res = self._tests.to_list()
s_res.sort(key=lambda user: (user.address.street, user.address.nr))
self.assertEqual(res, s_res)
def test_then_by_descending(self):
res = self._tests.order_by_descending(lambda user: user.address.street[0]).then_by_descending(
lambda user: user.address.nr)
res = self._tests.order_by_descending(lambda user: user.address.street).then_by_descending(lambda user: user.address.nr).to_list()
s_res = self._tests
s_res.sort(key=lambda user: (user.address.street[0], user.address.nr), reverse=True)
s_res = self._tests.to_list()
s_res.sort(key=lambda user: (user.address.street, user.address.nr), reverse=True)
self.assertEqual(res, s_res)
@ -237,7 +248,7 @@ class QueryTestCase(unittest.TestCase):
l_res = self._tests.to_list()
l_res.reverse()
self.assertEqual(l_res, res)
self.assertEqual(res.to_list(), l_res)
def test_select(self):
range_list = List(int, range(0, 100))
@ -251,10 +262,10 @@ class QueryTestCase(unittest.TestCase):
self.assertEqual(range_list.where(lambda x: x % 2 == 0).to_list(), modulo_range)
def test_select_many(self):
range_list = List(int, range(0, 100))
range_list = List(int, list(range(0, 100)))
selected_range = range_list.select(lambda x: [x, x])
self.assertEqual(selected_range, [[x, x] for x in range(0, 100)])
self.assertEqual(selected_range.to_list(), [[x, x] for x in range(0, 100)])
self.assertEqual(selected_range.select_many(lambda x: x).to_list(), [_x for _l in [2 * [x] for x in range(0, 100)] for _x in _l])
class TestClass:
@ -285,17 +296,17 @@ class QueryTestCase(unittest.TestCase):
self.assertIsNone(sn_res)
def test_skip(self):
skipped = self._tests.skip(5)
skipped = self._tests.skip(5).to_list()
self.assertEqual(len(self._tests) - 5, len(skipped))
self.assertEqual(self._tests[5:], skipped)
self.assertEqual(len(skipped), len(self._tests) - 5)
self.assertEqual(skipped, self._tests[5:])
def test_skip_last(self):
skipped = self._tests.skip_last(5)
self.assertEqual(len(self._tests) - 5, len(skipped))
self.assertEqual(self._tests[:-5], skipped)
self.assertEqual(self._tests[:-5][len(self._tests[:-5]) - 1], skipped.last())
self.assertEqual(skipped.count(), len(self._tests) - 5)
self.assertEqual(skipped.to_list(), self._tests[:-5])
self.assertEqual(skipped.last(), self._tests[:-5][len(self._tests[:-5]) - 1])
def test_sum(self):
res = self._tests.sum(lambda u: u.address.nr)
@ -318,15 +329,15 @@ class QueryTestCase(unittest.TestCase):
def test_take(self):
skipped = self._tests.take(5)
self.assertEqual(5, len(skipped))
self.assertEqual(self._tests[:5], skipped)
self.assertEqual(skipped.count(), 5)
self.assertEqual(skipped.to_list(), self._tests[:5])
def test_take_last(self):
skipped = self._tests.take_last(5)
self.assertEqual(5, len(skipped))
self.assertEqual(self._tests[-5:], skipped)
self.assertEqual(self._tests[len(self._tests) - 1], skipped.last())
self.assertEqual(skipped.count(), 5)
self.assertEqual(skipped.to_list(), self._tests[-5:])
self.assertEqual(skipped.last(), self._tests[len(self._tests) - 1])
def test_where(self):
results = []

16
unittests/unittests_query/iterable_test_case.py
View File

@ -17,5 +17,19 @@ class IterableTestCase(unittest.TestCase):
self._list.append(2)
self._list.append(3)
self.assertEqual(self._list, [1, 2, 3])
self.assertEqual(self._list.to_list(), [1, 2, 3])
self.assertRaises(Exception, lambda v: self._list.append(v), '3')
def test_assign(self):
self._list.append(1)
self._list.append(2)
self._list.append(3)
self._list[0] = 42
self.assertEqual(self._list[0], 42)
self._list[0] = 1
self._list.append(42)
self.assertEqual(self._list[3], 42)
del self._list[3]
self.assertEqual(self._list.to_list(), [1, 2, 3])
self.assertRaises(Exception, lambda v: self._list.append(v), '3')

68
unittests/unittests_query/performance_test_case.py Normal file
View File

@ -0,0 +1,68 @@
import sys
import timeit
import unittest
from cpl_query.enumerable import Enumerable
from cpl_query.iterable import Iterable
VALUES = 1000
COUNT = 75
class PerformanceTestCase(unittest.TestCase):
def setUp(self):
i = 0
self.values = []
while i < VALUES:
self.values.append(i)
i += 1
def test_range(self):
default = timeit.timeit(lambda: list(self.values), number=COUNT)
enumerable = timeit.timeit(lambda: Enumerable(int, self.values), number=COUNT)
iterable = timeit.timeit(lambda: Iterable(int, self.values), number=COUNT)
print('Range')
print(f'd: {default}s')
print(f'i: {iterable}s')
print(f'e: {enumerable}s')
self.assertLess(default, enumerable)
self.assertLess(default, iterable)
def test_where_single(self):
default = timeit.timeit(lambda: [x for x in list(self.values) if x == 50], number=COUNT)
iterable = timeit.timeit(lambda: Iterable(int, self.values).where(lambda x: x == 50).single(), number=COUNT)
enumerable = timeit.timeit(lambda: Enumerable(int, self.values).where(lambda x: x == 50).single(), number=COUNT)
print('Where single')
print(f'd: {default}s')
print(f'i: {iterable}s')
print(f'e: {enumerable}s')
self.assertLess(default, enumerable)
self.assertLess(default, iterable)
def test_where_single_complex(self):
class TestModel:
def __init__(self, v, tm=None):
self.value = v
self.tm = tm
values = []
for i in range(VALUES):
values.append(TestModel(i, TestModel(i + 1)))
default = timeit.timeit(lambda: [x for x in list(values) if x.tm.value == 50], number=COUNT)
iterable = timeit.timeit(lambda: Iterable(TestModel, values).where(lambda x: x.tm.value == 50).single(), number=COUNT)
enumerable = timeit.timeit(lambda: Enumerable(TestModel, values).where(lambda x: x.tm.value == 50).single(), number=COUNT)
print('Complex where single')
print(f'd: {default}s')
print(f'i: {iterable}s')
print(f'e: {enumerable}s')
self.assertLess(default, enumerable)
self.assertLess(default, iterable)

15
unittests/unittests_query/query_test_suite.py
View File

@ -1,7 +1,10 @@
import unittest
from unittests_query.enumerable_query_test_case import EnumerableQueryTestCase
from unittests_query.enumerable_test_case import EnumerableTestCase
from unittests_query.iterable_query_test_case import IterableQueryTestCase
from unittests_query.iterable_test_case import IterableTestCase
from unittests_query.query_test_case import QueryTestCase
from unittests_query.sequence_test_case import SequenceTestCase
class QueryTestSuite(unittest.TestSuite):
@ -10,8 +13,16 @@ class QueryTestSuite(unittest.TestSuite):
unittest.TestSuite.__init__(self)
loader = unittest.TestLoader()
self.addTests(loader.loadTestsFromTestCase(QueryTestCase))
self.addTests(loader.loadTestsFromTestCase(SequenceTestCase))
self.addTests(loader.loadTestsFromTestCase(EnumerableTestCase))
self.addTests(loader.loadTestsFromTestCase(EnumerableQueryTestCase))
self.addTests(loader.loadTestsFromTestCase(IterableTestCase))
self.addTests(loader.loadTestsFromTestCase(IterableQueryTestCase))
def run(self, *args):
super().run(*args)
if __name__ == "__main__":
runner = unittest.TextTestRunner()
runner.run(QueryTestSuite())

32
unittests/unittests_query/sequence_test_case.py Normal file
View File

@ -0,0 +1,32 @@
import unittest
from cpl_query.enumerable import Enumerable
from cpl_query.extension.list import List
from cpl_query.iterable import Iterable
class SequenceTestCase(unittest.TestCase):
def test_to_list(self):
_list = List().extend(range(0, 100))
enumerable = Enumerable.range(0, 100)
iterable = Iterable(int, list(range(0, 100)))
self.assertEqual(enumerable.to_list(), _list.to_list())
self.assertEqual(iterable.to_list(), _list.to_list())
def test_to_enumerable(self):
_list = List().extend(range(0, 100))
enumerable = Enumerable.range(0, 100)
iterable = Iterable(int, list(range(0, 100)))
self.assertEqual(type(_list.to_enumerable()), type(enumerable))
self.assertEqual(type(iterable.to_enumerable()), type(enumerable))
def test_to_iterable(self):
_list = List().extend(range(0, 100))
enumerable = Enumerable.range(0, 100)
iterable = Iterable(int, list(range(0, 100)))
self.assertEqual(type(_list.to_iterable()), type(iterable))
self.assertEqual(type(enumerable.to_iterable()), type(iterable))