Added cpl-mail
Some checks failed
Build on push / prepare (push) Successful in 9s
Build on push / core (push) Successful in 18s
Build on push / query (push) Successful in 25s
Build on push / translation (push) Failing after 8s
Build on push / mail (push) Successful in 14s

This commit is contained in:
2025-09-15 20:56:07 +02:00
parent 3b120370b8
commit 25b4ca0696
344 changed files with 4567 additions and 4946 deletions

View File

@@ -0,0 +1 @@

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,5 @@
from .default_lambda import default_lambda
from .ordered_queryable import OrderedQueryable
from .sequence import Sequence
from .ordered_queryable_abc import OrderedQueryableABC
from .queryable_abc import QueryableABC

View File

@@ -0,0 +1,2 @@
def default_lambda(x: object):
return x

View File

@@ -0,0 +1,34 @@
from collections.abc import Callable
from cpl.query.base.ordered_queryable_abc import OrderedQueryableABC
from cpl.query.exceptions import ArgumentNoneException, ExceptionArgument
class OrderedQueryable(OrderedQueryableABC):
r"""Implementation of :class: `cpl.query.base.ordered_queryable_abc.OrderedQueryableABC`"""
def __init__(self, _t: type, _values: OrderedQueryableABC = None, _func: Callable = None):
OrderedQueryableABC.__init__(self, _t, _values, _func)
def then_by(self, _func: Callable) -> OrderedQueryableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
self._funcs.append(_func)
return OrderedQueryable(self.type, sorted(self, key=lambda *args: [f(*args) for f in self._funcs]), _func)
def then_by_descending(self, _func: Callable) -> OrderedQueryableABC:
if self is None:
raise ArgumentNoneException(ExceptionArgument.list)
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
self._funcs.append(_func)
return OrderedQueryable(
self.type, sorted(self, key=lambda *args: [f(*args) for f in self._funcs], reverse=True), _func
)

View File

@@ -0,0 +1,38 @@
from __future__ import annotations
from abc import abstractmethod
from collections.abc import Callable
from typing import Iterable
from cpl.query.base.queryable_abc import QueryableABC
class OrderedQueryableABC(QueryableABC):
@abstractmethod
def __init__(self, _t: type, _values: Iterable = None, _func: Callable = None):
QueryableABC.__init__(self, _t, _values)
self._funcs: list[Callable] = []
if _func is not None:
self._funcs.append(_func)
@abstractmethod
def then_by(self, func: Callable) -> OrderedQueryableABC:
r"""Sorts OrderedList in ascending order by function
Parameter:
func: :class:`Callable`
Returns:
list of :class:`cpl.query.base.ordered_queryable_abc.OrderedQueryableABC`
"""
@abstractmethod
def then_by_descending(self, func: Callable) -> OrderedQueryableABC:
r"""Sorts OrderedList in descending order by function
Parameter:
func: :class:`Callable`
Returns:
list of :class:`cpl.query.base.ordered_queryable_abc.OrderedQueryableABC`
"""

View File

@@ -0,0 +1,569 @@
from __future__ import annotations
from typing import Optional, Callable, Union, Iterable, Any
from cpl.query._helper import is_number
from cpl.query.base import default_lambda
from cpl.query.base.sequence import Sequence
from cpl.query.exceptions import (
InvalidTypeException,
ArgumentNoneException,
ExceptionArgument,
IndexOutOfRangeException,
)
class QueryableABC(Sequence):
def __init__(self, t: type, values: Iterable = None):
Sequence.__init__(self, t, values)
def all(self, _func: Callable = None) -> bool:
r"""Checks if every element of list equals result found by function
Parameter
---------
func: :class:`Callable`
selected value
Returns
-------
bool
"""
if _func is None:
_func = default_lambda
return self.count(_func) == self.count()
def any(self, _func: Callable = None) -> bool:
r"""Checks if list contains result found by function
Parameter
---------
func: :class:`Callable`
selected value
Returns
-------
bool
"""
if _func is None:
_func = default_lambda
return self.where(_func).count() > 0
def average(self, _func: Callable = None) -> Union[int, float, complex]:
r"""Returns average value of list
Parameter
---------
func: :class:`Callable`
selected value
Returns
-------
Union[int, float, complex]
"""
if _func is None and not is_number(self.type):
raise InvalidTypeException()
return self.sum(_func) / self.count()
def contains(self, _value: object) -> bool:
r"""Checks if list contains value given by function
Parameter
---------
value: :class:`object`
value
Returns
-------
bool
"""
if _value is None:
raise ArgumentNoneException(ExceptionArgument.value)
return self.where(lambda x: x == _value).count() > 0
def count(self, _func: Callable = None) -> int:
r"""Returns length of list or count of found elements
Parameter
---------
func: :class:`Callable`
selected value
Returns
-------
int
"""
if _func is None:
return self.__len__()
return self.where(_func).count()
def distinct(self, _func: Callable = None) -> QueryableABC:
r"""Returns list without redundancies
Parameter
---------
func: :class:`Callable`
selected value
Returns
-------
:class: `cpl.query.base.queryable_abc.QueryableABC`
"""
if _func is None:
_func = default_lambda
result = []
known_values = []
for element in self:
value = _func(element)
if value in known_values:
continue
known_values.append(value)
result.append(element)
return type(self)(self._type, result)
def element_at(self, _index: int) -> any:
r"""Returns element at given index
Parameter
---------
_index: :class:`int`
index
Returns
-------
Value at _index: any
"""
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
if _index < 0 or _index >= self.count():
raise IndexOutOfRangeException
result = self._values[_index]
if result is None:
raise IndexOutOfRangeException
return result
def element_at_or_default(self, _index: int) -> Optional[any]:
r"""Returns element at given index or None
Parameter
---------
_index: :class:`int`
index
Returns
-------
Value at _index: Optional[any]
"""
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
try:
return self._values[_index]
except IndexError:
return None
def first(self) -> any:
r"""Returns first element
Returns
-------
First element of list: any
"""
if self.count() == 0:
raise IndexOutOfRangeException()
return self._values[0]
def first_or_default(self) -> any:
r"""Returns first element or None
Returns
-------
First element of list: Optional[any]
"""
if self.count() == 0:
return None
return self._values[0]
def for_each(self, _func: Callable = None):
r"""Runs given function for each element of list
Parameter
---------
func: :class: `Callable`
function to call
"""
if _func is not None:
for element in self:
_func(element)
return self
def group_by(self, _func: Callable = None) -> QueryableABC:
r"""Groups by func
Returns
-------
Grouped list[list[any]]: any
"""
if _func is None:
_func = default_lambda
groups = {}
for v in self:
value = _func(v)
if v not in groups:
groups[value] = []
groups[value].append(v)
v = []
for g in groups.values():
v.append(type(self)(object, g))
x = type(self)(type(self), v)
return x
def last(self) -> any:
r"""Returns last element
Returns
-------
Last element of list: any
"""
if self.count() == 0:
raise IndexOutOfRangeException()
return self._values[self.count() - 1]
def last_or_default(self) -> any:
r"""Returns last element or None
Returns
-------
Last element of list: Optional[any]
"""
if self.count() == 0:
return None
return self._values[self.count() - 1]
def max(self, _func: Callable = None) -> object:
r"""Returns the highest value
Parameter
---------
func: :class:`Callable`
selected value
Returns
-------
object
"""
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]:
r"""Return the median value of data elements
Returns
-------
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) -> object:
r"""Returns the lowest value
Parameter
---------
func: :class:`Callable`
selected value
Returns
-------
object
"""
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) -> "OrderedQueryableABC":
r"""Sorts elements by function in ascending order
Parameter
---------
func: :class:`Callable`
selected value
Returns
-------
:class: `cpl.query.base.ordered_queryable_abc.OrderedQueryableABC`
"""
if _func is None:
_func = default_lambda
from cpl.query.base.ordered_queryable import OrderedQueryable
return OrderedQueryable(self.type, sorted(self, key=_func), _func)
def order_by_descending(self, _func: Callable = None) -> "OrderedQueryableABC":
r"""Sorts elements by function in descending order
Parameter
---------
func: :class:`Callable`
selected value
Returns
-------
:class: `cpl.query.base.ordered_queryable_abc.OrderedQueryableABC`
"""
if _func is None:
_func = default_lambda
from cpl.query.base.ordered_queryable import OrderedQueryable
return OrderedQueryable(self.type, sorted(self, key=_func, reverse=True), _func)
def reverse(self) -> QueryableABC:
r"""Reverses list
Returns
-------
:class: `cpl.query.base.queryable_abc.QueryableABC`
"""
return type(self)(self._type, reversed(self._values))
def select(self, _func: Callable) -> QueryableABC:
r"""Formats each element of list to a given format
Returns
-------
:class: `cpl.query.base.queryable_abc.QueryableABC`
"""
if _func is None:
_func = default_lambda
_l = [_func(_o) for _o in self]
_t = type(_l[0]) if len(_l) > 0 else Any
return type(self)(_t, _l)
def select_many(self, _func: Callable) -> QueryableABC:
r"""Flattens resulting lists to one
Returns
-------
:class: `cpl.query.base.queryable_abc.QueryableABC`
"""
# The line below is pain. I don't understand anything of it...
# written on 09.11.2022 by Sven Heidemann
return type(self)(object, [_a for _o in self for _a in _func(_o)])
def single(self) -> any:
r"""Returns one single element of list
Returns
-------
Found value: any
Raises
------
ArgumentNoneException: when argument is None
Exception: when argument is None or found more than one element
"""
if self.count() > 1:
raise Exception("Found more than one element")
elif self.count() == 0:
raise Exception("Found no element")
return self._values[0]
def single_or_default(self) -> Optional[any]:
r"""Returns one single element of list
Returns
-------
Found value: Optional[any]
"""
if self.count() > 1:
raise Exception("Index out of range")
elif self.count() == 0:
return None
return self._values[0]
def skip(self, _index: int) -> QueryableABC:
r"""Skips all elements from index
Parameter
---------
_index: :class:`int`
index
Returns
-------
:class: `cpl.query.base.queryable_abc.QueryableABC`
"""
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
return type(self)(self.type, self._values[_index:])
def skip_last(self, _index: int) -> QueryableABC:
r"""Skips all elements after index
Parameter
---------
_index: :class:`int`
index
Returns
-------
:class: `cpl.query.base.queryable_abc.QueryableABC`
"""
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
index = self.count() - _index
return type(self)(self._type, self._values[:index])
def sum(self, _func: Callable = None) -> Union[int, float, complex]:
r"""Sum of all values
Parameter
---------
func: :class:`Callable`
selected value
Returns
-------
Union[int, float, complex]
"""
if _func is None and not is_number(self.type):
raise InvalidTypeException()
if _func is None:
_func = default_lambda
result = 0
for x in self:
result += _func(x)
return result
def split(self, _func: Callable) -> QueryableABC:
r"""Splits the list by given function
Parameter
---------
func: :class:`Callable`
seperator
Returns
-------
:class: `cpl.query.base.queryable_abc.QueryableABC`
"""
groups = []
group = []
for x in self:
v = _func(x)
if x == v:
groups.append(group)
group = []
group.append(x)
groups.append(group)
query_groups = []
for g in groups:
if len(g) == 0:
continue
query_groups.append(type(self)(self._type, g))
return type(self)(self._type, query_groups)
def take(self, _index: int) -> QueryableABC:
r"""Takes all elements from index
Parameter
---------
_index: :class:`int`
index
Returns
-------
:class: `cpl.query.base.queryable_abc.QueryableABC`
"""
if _index is None:
raise ArgumentNoneException(ExceptionArgument.index)
return type(self)(self._type, self._values[:_index])
def take_last(self, _index: int) -> QueryableABC:
r"""Takes all elements after index
Parameter
---------
_index: :class:`int`
index
Returns
-------
:class: `cpl.query.base.queryable_abc.QueryableABC`
"""
index = self.count() - _index
if index >= self.count() or index < 0:
raise IndexOutOfRangeException()
return type(self)(self._type, self._values[index:])
def where(self, _func: Callable = None) -> QueryableABC:
r"""Select element by function
Parameter
---------
func: :class:`Callable`
selected value
Returns
-------
:class: `cpl.query.base.queryable_abc.QueryableABC`
"""
if _func is None:
raise ArgumentNoneException(ExceptionArgument.func)
if _func is None:
_func = default_lambda
return type(self)(self.type, filter(_func, self))

View File

@@ -0,0 +1,96 @@
from abc import abstractmethod, ABC
from typing import Iterable
class Sequence(ABC):
@abstractmethod
def __init__(self, t: type, values: Iterable = None):
assert t is not None
assert isinstance(t, type) or t == any
assert values is None or isinstance(values, Iterable)
if values is None:
values = []
self._values = list(values)
if t is None:
t = object
self._type = t
def __iter__(self):
return iter(self._values)
def __next__(self):
return next(iter(self._values))
def __len__(self):
return self.to_list().__len__()
@classmethod
def __class_getitem__(cls, _t: type) -> type:
return _t
def __repr__(self):
return f"<{type(self).__name__} {self.to_list().__repr__()}>"
@property
def type(self) -> type:
return self._type
def _check_type(self, __object: any):
if self._type == any:
return
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}")
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._values]
def copy(self) -> "Sequence":
r"""Creates a copy of sequence
Returns:
Sequence
"""
return type(self)(self._type, self.to_list())
@classmethod
def empty(cls) -> "Sequence":
r"""Returns an empty sequence
Returns:
Sequence object that contains no elements
"""
return cls(object, [])
def index_of(self, _object: object) -> int:
r"""Returns the index of given element
Returns:
Index of object
Raises:
IndexError if object not in sequence
"""
for i, o in enumerate(self):
if o == _object:
return i
raise IndexError
@classmethod
def range(cls, start: int, length: int) -> "Sequence":
return cls(int, range(start, length))

View File

@@ -0,0 +1,2 @@
from .enumerable import Enumerable
from .enumerable_abc import EnumerableABC

View File

@@ -0,0 +1,12 @@
from cpl.query.enumerable.enumerable_abc import EnumerableABC
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: list = None):
EnumerableABC.__init__(self, t, values)

View File

@@ -0,0 +1,21 @@
from abc import abstractmethod
from cpl.query.base.queryable_abc import QueryableABC
class EnumerableABC(QueryableABC):
r"""ABC to define functions on list"""
@abstractmethod
def __init__(self, t: type = None, values: list = None):
QueryableABC.__init__(self, t, values)
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())

View File

@@ -0,0 +1,33 @@
from enum import Enum
# models
class ExceptionArgument(Enum):
list = "list"
func = "func"
type = "type"
value = "value"
index = "index"
# exceptions
class ArgumentNoneException(Exception):
r"""Exception when argument is None"""
def __init__(self, arg: ExceptionArgument):
Exception.__init__(self, f"argument {arg} is None")
class IndexOutOfRangeException(Exception):
r"""Exception when index is 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):
r"""Exception when type is invalid"""
class WrongTypeException(Exception):
r"""Exception when type is unexpected"""

View File

@@ -0,0 +1 @@
from .list import List

View File

@@ -0,0 +1,36 @@
from cpl.query.enumerable.enumerable_abc import EnumerableABC
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: Iterable = None):
Iterable.__init__(self, t, values)
def __getitem__(self, *args):
return self._values.__getitem__(*args)
def __setitem__(self, *args):
self._values.__setitem__(*args)
def __delitem__(self, *args):
self._values.__delitem__(*args)
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) -> Iterable:
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`
"""
return Iterable(self._type, self.to_list())

View File

@@ -0,0 +1,2 @@
from .iterable_abc import IterableABC
from .iterable import Iterable

View File

@@ -0,0 +1,8 @@
from typing import Iterable as TIterable
from cpl.query.iterable.iterable_abc import IterableABC
class Iterable(IterableABC):
def __init__(self, t: type = None, values: TIterable = None):
IterableABC.__init__(self, t, values)

View File

@@ -0,0 +1,71 @@
from abc import abstractmethod
from typing import Iterable
from cpl.query.base.queryable_abc import QueryableABC
class IterableABC(QueryableABC):
r"""ABC to define functions on list"""
@abstractmethod
def __init__(self, t: type = None, values: Iterable = None):
QueryableABC.__init__(self, t, values)
def __str__(self):
return str(self.to_list())
def append(self, _object: object):
self.add(_object)
def add(self, _object: object):
r"""Adds element to list
Parameter:
_object: :class:`object`
value
"""
self._check_type(_object)
self._values.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 remove(self, _object: object):
r"""Removes element from list
Parameter:
_object: :class:`object`
value
"""
if _object not in self:
raise ValueError
self._values.remove(_object)
def remove_at(self, _index: int):
r"""Removes element from list
Parameter:
_object: :class:`object`
value
"""
self._values.pop(_index)
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())

View File

@@ -0,0 +1,30 @@
[build-system]
requires = ["setuptools>=70.1.0", "wheel>=0.43.0"]
build-backend = "setuptools.build_meta"
[project]
name = "cpl-query"
version = "2024.7.0"
description = "CPL query"
readme ="CPL query package"
requires-python = ">=3.12"
license = { text = "MIT" }
authors = [
{ name = "Sven Heidemann", email = "sven.heidemann@sh-edraft.de" }
]
keywords = ["cpl", "query", "backend", "shared", "library"]
dynamic = ["dependencies", "optional-dependencies"]
[project.urls]
Homepage = "https://www.sh-edraft.de"
[tool.setuptools.packages.find]
where = ["."]
include = ["cpl*"]
[tool.setuptools.dynamic]
dependencies = { file = ["requirements.txt"] }
optional-dependencies.dev = { file = ["requirements.dev.txt"] }

View File

@@ -0,0 +1 @@
black==25.1.0

View File