2022.10.8 - Moved query functions to static query class (#102) #104
@ -1,26 +0,0 @@
|
||||
# -*- 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._query'
|
||||
__author__ = 'Sven Heidemann'
|
||||
__license__ = 'MIT'
|
||||
__copyright__ = 'Copyright (c) 2021 - 2022 sh-edraft.de'
|
||||
__version__ = '2022.10.2'
|
||||
|
||||
from collections import namedtuple
|
||||
|
||||
|
||||
# imports:
|
||||
|
||||
VersionInfo = namedtuple('VersionInfo', 'major minor micro')
|
||||
version_info = VersionInfo(major='2022', minor='10', micro='2')
|
||||
@ -1,16 +0,0 @@
|
||||
from collections.abc import Callable
|
||||
|
||||
from cpl_query._query.where import where_query
|
||||
from cpl_query.exceptions import ExceptionArgument, ArgumentNoneException
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def all_query(_list: IterableABC, _func: Callable) -> bool:
|
||||
if _list is None:
|
||||
raise ArgumentNoneException(ExceptionArgument.list)
|
||||
|
||||
if _func is None:
|
||||
raise ArgumentNoneException(ExceptionArgument.func)
|
||||
|
||||
result = where_query(_list, _func)
|
||||
return len(result) == len(_list)
|
||||
@ -1,16 +0,0 @@
|
||||
from collections.abc import Callable
|
||||
|
||||
from cpl_query._query.where import where_query
|
||||
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def any_query(_list: IterableABC, _func: Callable) -> bool:
|
||||
if _list is None:
|
||||
raise ArgumentNoneException(ExceptionArgument.list)
|
||||
|
||||
if _func is None:
|
||||
raise ArgumentNoneException(ExceptionArgument.func)
|
||||
|
||||
result = where_query(_list, _func)
|
||||
return len(result) > 0
|
||||
@ -1,27 +0,0 @@
|
||||
from typing import Callable, Union
|
||||
|
||||
from cpl_query._helper import is_number
|
||||
from cpl_query.exceptions import InvalidTypeException, WrongTypeException, ExceptionArgument, ArgumentNoneException
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def avg_query(_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
|
||||
@ -1,12 +0,0 @@
|
||||
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def contains_query(_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
|
||||
@ -1,15 +0,0 @@
|
||||
from collections.abc import Callable
|
||||
|
||||
from cpl_query._query.where import where_query
|
||||
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def count_query(_list: IterableABC, _func: Callable = None) -> int:
|
||||
if _list is None:
|
||||
raise ArgumentNoneException(ExceptionArgument.list)
|
||||
|
||||
if _func is None:
|
||||
return len(_list)
|
||||
|
||||
return len(where_query(_list, _func))
|
||||
@ -1,24 +0,0 @@
|
||||
from collections.abc import Callable
|
||||
|
||||
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def distinct_query(_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
|
||||
@ -1,25 +0,0 @@
|
||||
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def element_at_query(_list: IterableABC, _index: int) -> any:
|
||||
if _list is None:
|
||||
raise ArgumentNoneException(ExceptionArgument.list)
|
||||
|
||||
if _index is None:
|
||||
raise ArgumentNoneException(ExceptionArgument.index)
|
||||
|
||||
return _list[_index]
|
||||
|
||||
|
||||
def element_at_or_default_query(_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
|
||||
@ -1,44 +0,0 @@
|
||||
from typing import Optional
|
||||
|
||||
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument, IndexOutOfRangeException
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def first_query(_list: IterableABC) -> any:
|
||||
if _list is None:
|
||||
raise ArgumentNoneException(ExceptionArgument.list)
|
||||
|
||||
if len(_list) == 0:
|
||||
raise IndexOutOfRangeException()
|
||||
|
||||
return _list[0]
|
||||
|
||||
|
||||
def first_or_default_query(_list: IterableABC) -> Optional[any]:
|
||||
if _list is None:
|
||||
raise ArgumentNoneException(ExceptionArgument.list)
|
||||
|
||||
if len(_list) == 0:
|
||||
return None
|
||||
|
||||
return _list[0]
|
||||
|
||||
|
||||
def last_query(_list: IterableABC) -> any:
|
||||
if _list is None:
|
||||
raise ArgumentNoneException(ExceptionArgument.list)
|
||||
|
||||
if len(_list) == 0:
|
||||
raise IndexOutOfRangeException()
|
||||
|
||||
return _list[len(_list) - 1]
|
||||
|
||||
|
||||
def last_or_default_query(_list: IterableABC) -> Optional[any]:
|
||||
if _list is None:
|
||||
raise ArgumentNoneException(ExceptionArgument.list)
|
||||
|
||||
if len(_list) == 0:
|
||||
return None
|
||||
|
||||
return _list[len(_list) - 1]
|
||||
@ -1,15 +0,0 @@
|
||||
from collections.abc import Callable
|
||||
|
||||
from cpl_query.exceptions import ExceptionArgument, ArgumentNoneException
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def for_each_query(_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)
|
||||
@ -1,51 +0,0 @@
|
||||
from collections.abc import Callable
|
||||
from typing import Union
|
||||
|
||||
from cpl_query._helper import is_number
|
||||
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument, InvalidTypeException, WrongTypeException
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def max_query(_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
|
||||
|
||||
|
||||
def min_query(_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
|
||||
@ -1,47 +0,0 @@
|
||||
from collections.abc import Callable
|
||||
|
||||
from cpl_query.exceptions import ExceptionArgument, ArgumentNoneException
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
from cpl_query.extension.ordered_iterable_abc import OrderedIterableABC
|
||||
|
||||
|
||||
def order_by_query(_list: IterableABC, _func: Callable) -> OrderedIterableABC:
|
||||
result = OrderedIterableABC(_list.type, _func)
|
||||
_list.sort(key=_func)
|
||||
result.extend(_list)
|
||||
return result
|
||||
|
||||
|
||||
def order_by_descending_query(_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
|
||||
|
||||
|
||||
def then_by_query(_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
|
||||
|
||||
|
||||
def then_by_descending_query(_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
|
||||
@ -1,15 +0,0 @@
|
||||
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def reverse_query(_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
|
||||
@ -1,19 +0,0 @@
|
||||
from typing import Callable
|
||||
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def select_query(_list: IterableABC, _f: Callable) -> any:
|
||||
result = IterableABC()
|
||||
result.extend(_f(_o) for _o in _list)
|
||||
return result
|
||||
|
||||
|
||||
def select_many_query(_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
|
||||
@ -1,28 +0,0 @@
|
||||
from typing import Optional
|
||||
|
||||
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def single_query(_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]
|
||||
|
||||
|
||||
def single_or_default_query(_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]
|
||||
@ -1,66 +0,0 @@
|
||||
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument, IndexOutOfRangeException
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def skip_query(_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
|
||||
|
||||
|
||||
def skip_last_query(_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
|
||||
|
||||
|
||||
def take_query(_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
|
||||
|
||||
|
||||
def take_last_query(_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
|
||||
@ -1,25 +0,0 @@
|
||||
from collections.abc import Callable
|
||||
from typing import Union
|
||||
|
||||
from cpl_query._helper import is_number
|
||||
from cpl_query.exceptions import ExceptionArgument, ArgumentNoneException, InvalidTypeException
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def sum_query(_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
|
||||
@ -1,19 +0,0 @@
|
||||
from collections.abc import Callable
|
||||
|
||||
from cpl_query.exceptions import ArgumentNoneException, ExceptionArgument
|
||||
from cpl_query.extension.iterable_abc import IterableABC
|
||||
|
||||
|
||||
def where_query(_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
|
||||
@ -1,27 +1,8 @@
|
||||
from typing import Callable, Optional, Union
|
||||
|
||||
from cpl_query._query.all import all_query
|
||||
from cpl_query._query.any import any_query
|
||||
from cpl_query._query.avg import avg_query
|
||||
from cpl_query._query.contains import contains_query
|
||||
from cpl_query._query.count import count_query
|
||||
from cpl_query._query.distinct import distinct_query
|
||||
from cpl_query._query.element_at import (element_at_or_default_query,
|
||||
element_at_query)
|
||||
from cpl_query._query.first_last import (first_or_default_query, first_query,
|
||||
last_or_default_query, last_query)
|
||||
from cpl_query._query.for_each import for_each_query
|
||||
from cpl_query._query.max_min import max_query, min_query
|
||||
from cpl_query._query.order_by import order_by_descending_query, order_by_query
|
||||
from cpl_query._query.reverse import reverse_query
|
||||
from cpl_query._query.select import select_query, select_many_query
|
||||
from cpl_query._query.single import single_or_default_query, single_query
|
||||
from cpl_query._query.skip_take import (skip_last_query, skip_query,
|
||||
take_last_query, take_query)
|
||||
from cpl_query._query.sum import sum_query
|
||||
from cpl_query._query.where import where_query
|
||||
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):
|
||||
@ -30,94 +11,94 @@ class Iterable(IterableABC):
|
||||
IterableABC.__init__(self, t, values)
|
||||
|
||||
def any(self, func: Callable) -> bool:
|
||||
return any_query(self, func)
|
||||
return Query.any(self, func)
|
||||
|
||||
def all(self, func: Callable) -> bool:
|
||||
return all_query(self, func)
|
||||
return Query.all(self, func)
|
||||
|
||||
def average(self, func: Callable = None) -> Union[int, float, complex]:
|
||||
return avg_query(self, func)
|
||||
return Query.avg(self, func)
|
||||
|
||||
def contains(self, value: object) -> bool:
|
||||
return contains_query(self, value)
|
||||
return Query.contains(self, value)
|
||||
|
||||
def count(self, func: Callable = None) -> int:
|
||||
return count_query(self, func)
|
||||
return Query.count(self, func)
|
||||
|
||||
def distinct(self, func: Callable = None) -> IterableABC:
|
||||
return self.__to_self(distinct_query(self, func))
|
||||
return self.__to_self(Query.distinct(self, func))
|
||||
|
||||
def element_at(self, index: int) -> any:
|
||||
return element_at_query(self, index)
|
||||
return Query.element_at(self, index)
|
||||
|
||||
def element_at_or_default(self, index: int) -> Optional[any]:
|
||||
return element_at_or_default_query(self, index)
|
||||
return Query.element_at_or_default(self, index)
|
||||
|
||||
def last(self) -> any:
|
||||
return last_query(self)
|
||||
return Query.last(self)
|
||||
|
||||
def last_or_default(self) -> Optional[any]:
|
||||
return last_or_default_query(self)
|
||||
return Query.last_or_default(self)
|
||||
|
||||
def first(self) -> any:
|
||||
return first_query(self)
|
||||
return Query.first(self)
|
||||
|
||||
def first_or_default(self) -> Optional[any]:
|
||||
return first_or_default_query(self)
|
||||
return Query.first_or_default(self)
|
||||
|
||||
def for_each(self, func: Callable):
|
||||
for_each_query(self, func)
|
||||
Query.for_each(self, func)
|
||||
|
||||
def max(self, func: Callable = None) -> Union[int, float, complex]:
|
||||
return max_query(self, func)
|
||||
return Query.max(self, func)
|
||||
|
||||
def min(self, func: Callable = None) -> Union[int, float, complex]:
|
||||
return min_query(self, func)
|
||||
return Query.min(self, func)
|
||||
|
||||
def order_by(self, func: Callable) -> OrderedIterableABC:
|
||||
res = order_by_query(self, func)
|
||||
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 = order_by_descending_query(self, func)
|
||||
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 reverse_query(self)
|
||||
return Query.reverse(self)
|
||||
|
||||
def single(self) -> any:
|
||||
return single_query(self)
|
||||
return Query.single(self)
|
||||
|
||||
def single_or_default(self) -> Optional[any]:
|
||||
return single_or_default_query(self)
|
||||
return Query.single_or_default(self)
|
||||
|
||||
def select(self, _f: Callable) -> IterableABC:
|
||||
return self.__to_self(select_query(self, _f))
|
||||
return self.__to_self(Query.select(self, _f))
|
||||
|
||||
def select_many(self, _f: Callable) -> IterableABC:
|
||||
return self.__to_self(select_many_query(self, _f))
|
||||
return self.__to_self(Query.select_many(self, _f))
|
||||
|
||||
def skip(self, index: int) -> IterableABC:
|
||||
return self.__to_self(skip_query(self, index))
|
||||
return self.__to_self(Query.skip(self, index))
|
||||
|
||||
def skip_last(self, index: int) -> IterableABC:
|
||||
return self.__to_self(skip_last_query(self, index))
|
||||
return self.__to_self(Query.skip_last(self, index))
|
||||
|
||||
def sum(self, func: Callable = None) -> Union[int, float, complex]:
|
||||
return sum_query(self, func)
|
||||
return Query.sum(self, func)
|
||||
|
||||
def take(self, index: int) -> IterableABC:
|
||||
return self.__to_self(take_query(self, index))
|
||||
return self.__to_self(Query.take(self, index))
|
||||
|
||||
def take_last(self, index: int) -> IterableABC:
|
||||
return self.__to_self(take_last_query(self, index))
|
||||
return self.__to_self(Query.take_last(self, index))
|
||||
|
||||
def where(self, func: Callable) -> IterableABC:
|
||||
return self.__to_self(where_query(self, func))
|
||||
return self.__to_self(Query.where(self, func))
|
||||
|
||||
@staticmethod
|
||||
def __to_self(obj: IterableABC) -> IterableABC:
|
||||
|
||||
@ -1,8 +1,8 @@
|
||||
from collections.abc import Callable
|
||||
|
||||
from cpl_query._query.order_by import then_by_descending_query, then_by_query
|
||||
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):
|
||||
@ -15,8 +15,8 @@ class OrderedIterable(Iterable, OrderedIterableABC):
|
||||
|
||||
def then_by(self, _func: Callable) -> OrderedIterableABC:
|
||||
self._funcs.append(_func)
|
||||
return then_by_query(self, lambda *args: [f(*args) for f in self._funcs])
|
||||
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 then_by_descending_query(self, lambda *args: [f(*args) for f in self._funcs])
|
||||
return Query.then_by_descending(self, lambda *args: [f(*args) for f in self._funcs])
|
||||
|
||||
405
src/cpl_query/query.py
Normal file
405
src/cpl_query/query.py
Normal file
@ -0,0 +1,405 @@
|
||||
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
|
||||
Loading…
Reference in New Issue
Block a user