2021.10.3 #35

Merged
edraft merged 31 commits from 2021.10.3 into 2021.10 2021-08-02 14:36:46 +02:00
39 changed files with 1307 additions and 14 deletions

View File

@ -3,9 +3,9 @@
"DefaultProject": "cpl_cli",
"Projects": {
"cpl": "src/cpl/cpl.json",
"cpl_cli": "src/cpl_cli/cpl_cli.json"
"cpl_cli": "src/cpl_cli/cpl_cli.json",
"cpl_query": "src/cpl_query/cpl_query.json"
},
"Scripts": {
}
"Scripts": {}
}
}

View File

@ -7,6 +7,10 @@ upload:
twine upload --repository-url https://pip.sh-edraft.de dist/sh_cpl-cli/publish/setup/*
twine upload -r pip.sh-edraft.de dist/sh_cpl-cli/publish/setup/*
query:
twine upload --repository-url https://pip-dev.sh-edraft.de dist/sh_cpl-query/publish/setup/*
twine upload -r pip-dev.sh-edraft.de dist/sh_cpl-query/publish/setup/*
exp:
cpl:
twine upload --repository-url https://pip-exp.sh-edraft.de dist/sh_cpl/publish/setup/*
@ -16,6 +20,10 @@ upload:
twine upload --repository-url https://pip-exp.sh-edraft.de dist/sh_cpl-cli/publish/setup/*
twine upload -r pip-exp.sh-edraft.de dist/sh_cpl-cli/publish/setup/*
query:
twine upload --repository-url https://pip-exp.sh-edraft.de dist/sh_cpl-query/publish/setup/*
twine upload -r pip-exp.sh-edraft.de dist/sh_cpl-query/publish/setup/*
dev:
cpl:
twine upload --repository-url https://pip-dev.sh-edraft.de dist/sh_cpl/publish/setup/*
@ -25,15 +33,22 @@ upload:
twine upload --repository-url https://pip-dev.sh-edraft.de dist/sh_cpl-cli/publish/setup/*
twine upload -r pip-dev.sh-edraft.de dist/sh_cpl-cli/publish/setup/*
query:
twine upload --repository-url https://pip-dev.sh-edraft.de dist/sh_cpl-query/publish/setup/*
twine upload -r pip-dev.sh-edraft.de dist/sh_cpl-query/publish/setup/*
install:
prod:
pip install --extra-index-url https://pip.sh-edraft.de/ sh_cpl
pip install --extra-index-url https://pip.sh-edraft.de/ sh_cpl-cli
pip install --extra-index-url https://pip.sh-edraft.de/ sh_cpl-query
exp:
pip install --extra-index-url https://pip-exp.sh-edraft.de/ sh_cpl
pip install --extra-index-url https://pip-exp.sh-edraft.de/ sh_cpl-cli
pip install --extra-index-url https://pip-exp.sh-edraft.de/ sh_cpl-query
dev:
pip install --extra-index-url https://pip-dev.sh-edraft.de/ sh_cpl
pip install --extra-index-url https://pip-dev.sh-edraft.de/ sh_cpl-cli
pip install --extra-index-url https://pip-dev.sh-edraft.de/ sh_cpl-query

View File

@ -2,10 +2,14 @@
# activate venv
source /home/sven/Nextcloud_Sven/Schreibtisch/git_sh-edraft_de/sh_cpl/cpl-env/bin/activate
# CLI
cd /home/sven/Nextcloud_Sven/Schreibtisch/git_sh-edraft_de/sh_cpl/
cpl build
# CPL
cd /home/sven/Nextcloud_Sven/Schreibtisch/git_sh-edraft_de/sh_cpl/src/cpl
cpl build
# CLI
cd /home/sven/Nextcloud_Sven/Schreibtisch/git_sh-edraft_de/sh_cpl/src/cpl_cli
cpl build
# CPL Query
cd /home/sven/Nextcloud_Sven/Schreibtisch/git_sh-edraft_de/sh_cpl/src/cpl_query
cpl build

View File

@ -2,10 +2,14 @@
# activate venv
source /home/sven/Nextcloud_Sven/Schreibtisch/git_sh-edraft_de/sh_cpl/cpl-env/bin/activate
# CLI
cd /home/sven/Nextcloud_Sven/Schreibtisch/git_sh-edraft_de/sh_cpl/
cpl publish
# CPL
cd /home/sven/Nextcloud_Sven/Schreibtisch/git_sh-edraft_de/sh_cpl/src/cpl
cpl publish
# CLI
cd /home/sven/Nextcloud_Sven/Schreibtisch/git_sh-edraft_de/sh_cpl/src/cpl_cli
cpl publish
# CPL Query
cd /home/sven/Nextcloud_Sven/Schreibtisch/git_sh-edraft_de/sh_cpl/src/cpl_query
cpl

View File

@ -43,7 +43,8 @@ class ConsoleBuilder:
ws_dict = {
WorkspaceSettings.__name__: {
WorkspaceSettingsNameEnum.default_project.value: project_name,
WorkspaceSettingsNameEnum.projects.value: projects
WorkspaceSettingsNameEnum.projects.value: projects,
WorkspaceSettingsNameEnum.scripts: {}
}
}

View File

@ -43,7 +43,8 @@ class LibraryBuilder:
ws_dict = {
WorkspaceSettings.__name__: {
WorkspaceSettingsNameEnum.default_project.value: project_name,
WorkspaceSettingsNameEnum.projects.value: projects
WorkspaceSettingsNameEnum.projects.value: projects,
WorkspaceSettingsNameEnum.scripts: {}
}
}

25
src/cpl_query/__init__.py Normal file
View File

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

View File

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

View File

@ -0,0 +1,115 @@
from typing import Optional, Callable, Union
from cpl_query.extension.ordered_iterable_abc import OrderedIterableABC
from .._query.all import all_query
from .._query.any import any_query
from .._query.avg import avg_query
from .._query.contains import contains_query
from .._query.count import count_query
from .._query.distinct import distinct_query
from .._query.element_at import element_at_query, element_at_or_default_query
from .._query.first_last import first_or_default_query, first_query, last_or_default_query, last_query
from .._query.for_each import for_each_query
from .._query.max_min import max_query, min_query
from .._query.order_by import order_by_query, order_by_descending_query
from .._query.reverse import reverse_query
from .._query.single import single_query, single_or_default_query
from .._query.skip_take import skip_query, skip_last_query, take_query, take_last_query
from .._query.sum import sum_query
from .._query.where import where_query
from cpl_query.extension.iterable_abc import IterableABC
class Iterable(IterableABC):
def __init__(self, t: type = None, values: list = None):
IterableABC.__init__(self, t, values)
def any(self, func: Callable) -> bool:
return any_query(self, func)
def all(self, func: Callable) -> bool:
return all_query(self, func)
def average(self, func: Callable = None) -> Union[int, float, complex]:
return avg_query(self, func)
def contains(self, value: object) -> bool:
return contains_query(self, value)
def count(self, func: Callable = None) -> int:
return count_query(self, func)
def distinct(self, func: Callable) -> IterableABC:
return self.__to_self(distinct_query(self, func))
def element_at(self, index: int) -> any:
return element_at_query(self, index)
def element_at_or_default(self, index: int) -> Optional[any]:
return element_at_or_default_query(self, index)
def last(self) -> any:
return last_query(self)
def last_or_default(self) -> Optional[any]:
return last_or_default_query(self)
def first(self) -> any:
return first_query(self)
def first_or_default(self) -> Optional[any]:
return first_or_default_query(self)
def for_each(self, func: Callable):
for_each_query(self, func)
def max(self, func: Callable = None) -> Union[int, float, complex]:
return max_query(self, func)
def min(self, func: Callable = None) -> Union[int, float, complex]:
return min_query(self, func)
def order_by(self, func: Callable) -> OrderedIterableABC:
res = order_by_query(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)
from cpl_query._extension.ordered_iterable import OrderedIterable
res.__class__ = OrderedIterable
return res
def reverse(self) -> IterableABC:
return reverse_query(self)
def single(self) -> any:
return single_query(self)
def single_or_default(self) -> Optional[any]:
return single_or_default_query(self)
def skip(self, index: int) -> IterableABC:
return self.__to_self(skip_query(self, index))
def skip_last(self, index: int) -> IterableABC:
return self.__to_self(skip_last_query(self, index))
def sum(self, func: Callable = None) -> Union[int, float, complex]:
return sum_query(self, func)
def take(self, index: int) -> IterableABC:
return self.__to_self(take_query(self, index))
def take_last(self, index: int) -> IterableABC:
return self.__to_self(take_last_query(self, index))
def where(self, func: Callable) -> IterableABC:
return self.__to_self(where_query(self, func))
@staticmethod
def __to_self(obj: IterableABC) -> IterableABC:
obj.__class__ = Iterable
return obj

View File

@ -0,0 +1,20 @@
from collections import Callable
from .iterable import Iterable
from .._query.order_by import then_by_query, then_by_descending_query
from cpl_query.extension.ordered_iterable_abc import OrderedIterableABC
class OrderedIterable(Iterable, OrderedIterableABC):
def __init__(self):
Iterable.__init__(self)
OrderedIterableABC.__init__(self)
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])
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])

2
src/cpl_query/_helper.py Normal file
View File

@ -0,0 +1,2 @@
def is_number(t: type) -> bool:
return issubclass(t, int) or issubclass(t, float) or issubclass(t, complex)

View File

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

View File

@ -0,0 +1,16 @@
from collections 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)

View File

@ -0,0 +1,16 @@
from collections 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

View File

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

View File

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

View File

@ -0,0 +1,15 @@
from collections 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))

View File

@ -0,0 +1,24 @@
from collections 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

View File

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

View File

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

View File

@ -0,0 +1,15 @@
from collections 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)

View File

@ -0,0 +1,51 @@
from collections 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

View File

@ -0,0 +1,47 @@
from collections 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(_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(_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

View File

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

View File

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

View File

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

View File

@ -0,0 +1,25 @@
from collections 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

View File

@ -0,0 +1,19 @@
from collections 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()
for element in _list:
if _func(element):
result.append(element)
return result

View File

@ -0,0 +1,41 @@
{
"ProjectSettings": {
"Name": "sh_cpl-query",
"Version": {
"Major": "2021",
"Minor": "10",
"Micro": "0.rc1"
},
"Author": "Sven Heidemann",
"AuthorEmail": "sven.heidemann@sh-edraft.de",
"Description": "sh-edraft Common Python library Query",
"LongDescription": "sh-edraft Common Python library Python integrated Queries",
"URL": "https://www.sh-edraft.de",
"CopyrightDate": "2020 - 2021",
"CopyrightName": "sh-edraft.de",
"LicenseName": "MIT",
"LicenseDescription": "MIT, see LICENSE for more details.",
"Dependencies": [
"sh_cpl==2021.4.0.post1"
],
"PythonVersion": ">=3.9.2",
"PythonPath": {},
"Classifiers": []
},
"BuildSettings": {
"ProjectType": "library",
"SourcePath": "",
"OutputPath": "../../dist",
"Main": "",
"EntryPoint": "",
"IncludePackageData": true,
"Included": [],
"Excluded": [
"*/__pycache__",
"*/logs",
"*/tests"
],
"PackageData": {},
"ProjectReferences": []
}
}

View File

@ -0,0 +1,31 @@
from enum import Enum
# models
class ExceptionArgument(Enum):
list = 'list'
func = 'func'
type = 'type'
value = 'value'
index = 'index'
# exceptions
class ArgumentNoneException(Exception):
def __init__(self, arg: ExceptionArgument):
Exception.__init__(self, f'argument {arg} is None')
class IndexOutOfRangeException(Exception):
def __init__(self):
Exception.__init__(self, f'List index out of range')
class InvalidTypeException(Exception):
pass
class WrongTypeException(Exception):
pass

View File

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

View File

@ -0,0 +1,115 @@
from abc import ABC, abstractmethod
from typing import Optional, Callable, Union, Iterable
class IterableABC(ABC, list):
@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 any(self, func: Callable) -> bool: pass
@abstractmethod
def all(self, func: Callable) -> bool: pass
def append(self, __object: object) -> None:
if self._type is not None and type(__object) != self._type and not isinstance(type(__object), self._type):
raise Exception(f'Unexpected type: {type(__object)}')
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]: pass
@abstractmethod
def contains(self, value: object) -> bool: pass
@abstractmethod
def count(self, func: Callable) -> int: pass
@abstractmethod
def distinct(self, func: Callable) -> 'IterableABC': pass
@abstractmethod
def element_at(self, index: int) -> any: pass
@abstractmethod
def element_at_or_default(self, index: int) -> Optional[any]: pass
def extend(self, __iterable: Iterable) -> None:
for value in __iterable:
self.append(value)
@abstractmethod
def last(self) -> any: pass
@abstractmethod
def last_or_default(self) -> any: pass
@abstractmethod
def first(self) -> any: pass
@abstractmethod
def first_or_default(self) -> any: pass
@abstractmethod
def for_each(self, func: Callable) -> Union[int, float, complex]: pass
@abstractmethod
def max(self, func: Callable = None) -> Union[int, float, complex]: pass
@abstractmethod
def min(self, func: Callable = None) -> Union[int, float, complex]: pass
@abstractmethod
def order_by(self, func: Callable) -> 'IterableABC': pass
@abstractmethod
def order_by_descending(self, func: Callable) -> 'IterableABC': pass
@abstractmethod
def reverse(self) -> 'IterableABC': pass
@abstractmethod
def single(self) -> any: pass
@abstractmethod
def single_or_default(self) -> Optional[any]: pass
@abstractmethod
def skip(self, index: int) -> 'IterableABC': pass
@abstractmethod
def skip_last(self, index: int) -> 'IterableABC': pass
@abstractmethod
def sum(self, func: Callable = None) -> Union[int, float, complex]: pass
@abstractmethod
def take(self, index: int) -> 'IterableABC': pass
@abstractmethod
def take_last(self, index: int) -> 'IterableABC': pass
def to_list(self) -> list:
return list(self)
@abstractmethod
def where(self, func: Callable) -> 'IterableABC': pass

View File

@ -0,0 +1,7 @@
from .._extension.iterable import Iterable
class List(Iterable):
def __init__(self, t: type = None, values: list = None):
Iterable.__init__(self, t, values)

View File

@ -0,0 +1,20 @@
from abc import abstractmethod
from collections import Callable
from cpl_query.extension.iterable_abc import IterableABC
class OrderedIterableABC(IterableABC):
@abstractmethod
def __init__(self, _func: Callable = None):
IterableABC.__init__(self)
self._funcs: list[Callable] = []
if _func is not None:
self._funcs.append(_func)
@abstractmethod
def then_by(self, func: Callable) -> 'OrderedIterableABC': pass
@abstractmethod
def then_by_descending(self, func: Callable) -> 'OrderedIterableABC': pass

View File

View File

@ -0,0 +1,21 @@
import unittest
from cpl_query.extension.list import List
class IterableTest(unittest.TestCase):
def setUp(self) -> None:
self._list = List(int)
def _clear(self):
self._list.clear()
self.assertEqual(self._list, [])
def test_append(self):
self._list.append(1)
self._list.append(2)
self._list.append(3)
self.assertEqual(self._list, [1, 2, 3])
self.assertRaises(Exception, lambda v: self._list.append(v), '3')

View File

@ -0,0 +1,12 @@
class User:
def __init__(self, name, address):
self.name = name
self.address = address
class Address:
def __init__(self, street, nr):
self.street = street
self.nr = nr

View File

@ -0,0 +1,314 @@
import string
import unittest
from random import randint
from cpl.utils import String
from cpl_query.exceptions import InvalidTypeException, ArgumentNoneException
from cpl_query.extension.list import List
from cpl_query.tests.models import User, Address
class QueryTest(unittest.TestCase):
def setUp(self) -> None:
self._tests = List(User)
self._t_user = User(
'Test user',
Address(
'teststr.',
15
)
)
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.append(user)
self._tests.append(self._t_user)
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 = List(str, ['hello', 'world'])
e_res = tests.average()
self.assertRaises(InvalidTypeException, invalid)
tests = List(int, list(range(0, 100)))
self.assertEqual(sum(tests) / len(tests), tests.average())
def wrong2():
tests2 = List(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[index], self._tests.element_at(index))
def test_element_at_or_default(self):
index = randint(0, len(self._tests) - 1)
self.assertEqual(self._tests[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[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[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[0], s_res)
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[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(self._t_user.address.nr, res)
tests = List(values=list(range(0, 100)))
self.assertEqual(99, tests.max())
def invalid():
tests = List(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 = List(values=list(range(0, 100)))
self.assertEqual(0, tests.min())
def invalid():
tests = List(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)
s_res = self._tests
s_res.sort(key=lambda user: user.address.street)
self.assertEqual(res, s_res)
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
s_res.sort(key=lambda user: user.address.street, reverse=True)
self.assertEqual(res, s_res)
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)
s_res = self._tests
s_res.sort(key=lambda user: (user.address.street[0], 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)
s_res = self._tests
s_res.sort(key=lambda user: (user.address.street[0], user.address.nr), reverse=True)
self.assertEqual(res, s_res)
def test_reverse(self):
res = self._tests.reverse()
l_res = self._tests.to_list()
l_res.reverse()
self.assertEqual(l_res, res)
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)
self.assertEqual(len(self._tests) - 5, len(skipped))
self.assertEqual(self._tests[5:], skipped)
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())
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 = List(values=list(range(0, 100)))
self.assertEqual(0, tests.min())
def invalid():
tests2 = List(str, ['hello', 'world'])
e_res = tests2.average()
self.assertRaises(InvalidTypeException, invalid)
def test_take(self):
skipped = self._tests.take(5)
self.assertEqual(5, len(skipped))
self.assertEqual(self._tests[:5], skipped)
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())
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)

View File

@ -0,0 +1,25 @@
import unittest
from cpl_query.tests.iterable_test import IterableTest
from cpl_query.tests.query_test import QueryTest
class Tester:
def __init__(self):
self._suite = unittest.TestSuite()
def create(self):
loader = unittest.TestLoader()
self._suite.addTests(loader.loadTestsFromTestCase(QueryTest))
self._suite.addTests(loader.loadTestsFromTestCase(IterableTest))
def start(self):
runner = unittest.TextTestRunner()
runner.run(self._suite)
if __name__ == '__main__':
tester = Tester()
tester.create()
tester.start()