Added order by functions

This commit is contained in:
Sven Heidemann 2021-07-26 15:21:57 +02:00
parent cc7755bafc
commit 0cae3428b9
7 changed files with 172 additions and 38 deletions

View File

@ -0,0 +1,28 @@
from collections import Callable
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:
result = OrderedIterableABC(_func)
_list.sort(key=_func, reverse=True)
result.extend(_list)
return result
def then_by_query(_list: OrderedIterableABC, _func: Callable) -> OrderedIterableABC:
_list.sort(key=_func)
return _list
def then_by_descending_query(_list: OrderedIterableABC, _func: Callable) -> OrderedIterableABC:
_list.sort(key=_func, reverse=True)
return _list

View File

@ -0,0 +1,53 @@
from typing import Optional, Callable
from .ordered_iterable import OrderedIterable
from .ordered_iterable_abc import OrderedIterableABC
from .._query.any_query import any_query
from .._query.first_query import first_or_default_query, first_query
from .._query.for_each_query import for_each_query
from .._query.order_by import order_by_query, order_by_descending_query
from .._query.single_query import single_query, single_or_default_query
from .._query.where_query import where_query
from cpl_query.extension.iterable_abc import IterableABC
class Iterable(IterableABC):
def __init__(self):
IterableABC.__init__(self)
def any(self, func: str) -> bool:
return any_query(self, func)
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 order_by(self, func: Callable) -> OrderedIterableABC:
res = order_by_query(self, func)
res.__class__ = OrderedIterable
return res
def order_by_descending(self, func: Callable) -> OrderedIterableABC:
res = order_by_descending_query(self, func)
res.__class__ = OrderedIterable
return res
def single(self) -> any:
return single_query(self)
def single_or_default(self) -> Optional[any]:
return single_or_default_query(self)
def where(self, func: str) -> IterableABC:
res = where_query(self, func)
res.__class__ = Iterable
return res
def to_list(self) -> list:
return list(self)

View File

@ -20,6 +20,12 @@ class IterableABC(ABC, list):
@abstractmethod @abstractmethod
def for_each(self, func: Callable): pass def for_each(self, func: Callable): pass
@abstractmethod
def order_by(self, func: Callable): pass
@abstractmethod
def order_by_descending(self, func: Callable): pass
@abstractmethod @abstractmethod
def single(self): pass def single(self): pass

View File

@ -1,40 +1,12 @@
from typing import Optional, Callable from cpl_query.extension.iterable import Iterable
from .._query.any_query import any_query
from .._query.first_query import first_or_default_query, first_query
from .._query.for_each_query import for_each_query
from .._query.single_query import single_query, single_or_default_query
from .._query.where_query import where_query
from cpl_query.extension.iterable_abc import IterableABC
class List(IterableABC): class List(Iterable):
def __init__(self): def __init__(self, t: type = None, values: list = None):
IterableABC.__init__(self) Iterable.__init__(self)
def any(self, func: str) -> bool: self._type = t
return any_query(self, func)
def first(self) -> any: if values is not None:
return first_query(self) self.extend(values)
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 single(self) -> any:
return single_query(self)
def single_or_default(self) -> Optional[any]:
return single_or_default_query(self)
def where(self, func: str) -> IterableABC:
res = where_query(self, func)
res.__class__ = List
return res
def to_list(self) -> list:
return list(self)

View File

@ -0,0 +1,19 @@
from abc import ABC
from collections import Callable
from .._query.order_by import then_by_query, then_by_descending_query
from cpl_query.extension.ordered_iterable_abc import OrderedIterableABC
class OrderedIterable(OrderedIterableABC, ABC):
def __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])

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

@ -10,7 +10,7 @@ from cpl_query.tests.models import User, Address
class QueryTest(unittest.TestCase): class QueryTest(unittest.TestCase):
def setUp(self) -> None: def setUp(self) -> None:
self._tests = List() self._tests = List(User)
self._t_user = User( self._t_user = User(
'Test user', 'Test user',
Address( Address(
@ -24,9 +24,9 @@ class QueryTest(unittest.TestCase):
def _generate_test_data(self): def _generate_test_data(self):
for i in range(0, 100): for i in range(0, 100):
user = User( user = User(
String.random_string(string.ascii_letters, 8), String.random_string(string.ascii_letters, 8).lower(),
Address( Address(
String.random_string(string.ascii_letters, 10), String.random_string(string.ascii_letters, 10).lower(),
randint(0, 10) randint(0, 10)
) )
) )
@ -85,6 +85,42 @@ class QueryTest(unittest.TestCase):
self.assertEqual(len(users), len(self._tests)) self.assertEqual(len(users), len(self._tests))
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)
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_single(self): def test_single(self):
res = self._tests.where(f'User.address.nr == {self._t_user.address.nr}') res = self._tests.where(f'User.address.nr == {self._t_user.address.nr}')
s_res = self._tests.where(f'User.address.nr == {self._t_user.address.nr}').single() s_res = self._tests.where(f'User.address.nr == {self._t_user.address.nr}').single()