From 38b8fd6df716fc6e78161e4ef6356f270fd8749a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Piotr=20Ma=C5=9Blanka?= <piotr.maslanka@henrietta.com.pl> Date: Sat, 19 Sep 2020 19:51:09 +0200 Subject: [PATCH] add loop_while and refactor the docs --- docs/coding/decorators.rst | 63 ++++++++++++ docs/coding/functions.rst | 64 ------------ docs/index.rst | 1 + satella/__init__.py | 2 +- satella/coding/decorators/__init__.py | 6 +- satella/coding/decorators/decorators.py | 77 +------------- satella/coding/decorators/flow_control.py | 120 ++++++++++++++++++++++ tests/test_coding/test_decorators.py | 25 ++++- 8 files changed, 214 insertions(+), 144 deletions(-) create mode 100644 docs/coding/decorators.rst create mode 100644 satella/coding/decorators/flow_control.py diff --git a/docs/coding/decorators.rst b/docs/coding/decorators.rst new file mode 100644 index 00000000..8260de5f --- /dev/null +++ b/docs/coding/decorators.rst @@ -0,0 +1,63 @@ +Decorators +========== + + +.. autofunction:: satella.coding.decorators.queue_get + +.. autofunction:: satella.coding.decorators.loop_while + +.. autofunction:: satella.coding.for_argument + +.. autofunction:: satella.coding.chain_functions + +.. autofunction:: satella.coding.auto_adapt_to_methods + +.. autofunction:: satella.coding.attach_arguments + +A ``functools.wraps()`` equivalent, but for classes + +.. autofunction:: satella.coding.wraps + +.. autofunction:: satella.coding.decorators.execute_before + + +Preconditions and postconditions +-------------------------------- + +Sometimes you need to specify conditions that parameter to your function will need to obey. +You can use the following decorator for this: + +.. autofunction:: satella.coding.precondition + +And here are some helper functions for it: + +has_keys asserts that a dictionary has all the keys necessary. + +.. autofunction:: satella.coding.has_keys + +Use it like this: + +>>> @precondition(has_keys(['a', 'b'])) +>>> def function(keys): +>>> ... +>>> function({'a': 5, 'b': 3}) +>>> self.assertRaises(PreconditionError, lambda: function({'a': 5})) + +short_none is particularly useful with preconditions, or functions +that accept a None value as well. + +.. autofunction:: satella.coding.short_none + +Example: + +>>> @precondition(short_none('x == 2')) +>>> def expect_two(x): +>>> ... +>>> expect_two(None) +>>> expect_two(2) +>>> self.assertRaises(PreconditionError, lambda: expect_two(3)) + +You can also check the return value with + +.. autofunction:: satella.coding.postcondition + diff --git a/docs/coding/functions.rst b/docs/coding/functions.rst index 323b8602..3d91d61d 100644 --- a/docs/coding/functions.rst +++ b/docs/coding/functions.rst @@ -25,12 +25,6 @@ Functions and decorators .. autofunction:: satella.coding.raises_exception -.. autofunction:: satella.coding.for_argument - -.. autofunction:: satella.coding.chain_functions - -.. autofunction:: satella.coding.auto_adapt_to_methods - Wrapping classes with something ------------------------------- @@ -95,61 +89,3 @@ Without running into `TypeError: metaclass conflict: the metaclass of a derived Following function will help with that: .. autofunction:: satella.coding.metaclass_maker - -Preconditions and postconditions --------------------------------- - -Sometimes you need to specify conditions that parameter to your function will need to obey. -You can use the following decorator for this: - -.. autofunction:: satella.coding.precondition - -And here are some helper functions for it: - -has_keys asserts that a dictionary has all the keys necessary. - -.. autofunction:: satella.coding.has_keys - -Use it like this: - ->>> @precondition(has_keys(['a', 'b'])) ->>> def function(keys): ->>> ... ->>> function({'a': 5, 'b': 3}) ->>> self.assertRaises(PreconditionError, lambda: function({'a': 5})) - -short_none is particularly useful with preconditions, or functions -that accept a None value as well. - -.. autofunction:: satella.coding.short_none - -Example: - ->>> @precondition(short_none('x == 2')) ->>> def expect_two(x): ->>> ... ->>> expect_two(None) ->>> expect_two(2) ->>> self.assertRaises(PreconditionError, lambda: expect_two(3)) - -You can also check the return value with - -.. autofunction:: satella.coding.postcondition - - -attach_arguments ----------------- - -.. autofunction:: satella.coding.attach_arguments - -wraps ------ -A ``functools.wraps()`` equivalent, but for classes - -.. autofunction:: satella.coding.wraps - -execute_before --------------- - -.. autofunction:: satella.coding.decorators.execute_before - diff --git a/docs/index.rst b/docs/index.rst index 7835d8eb..72ccf786 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -14,6 +14,7 @@ Visit the project's page at GitHub_! configuration/sources coding/functions coding/structures + coding/decorators coding/predicates coding/concurrent coding/sequences diff --git a/satella/__init__.py b/satella/__init__.py index b9be3b74..1cbd3c4d 100644 --- a/satella/__init__.py +++ b/satella/__init__.py @@ -1 +1 @@ -__version__ = '2.11.10_a3' +__version__ = '2.11.10_a4' diff --git a/satella/coding/decorators/__init__.py b/satella/coding/decorators/__init__.py index 898c8208..6db78f82 100644 --- a/satella/coding/decorators/__init__.py +++ b/satella/coding/decorators/__init__.py @@ -1,7 +1,9 @@ from .arguments import auto_adapt_to_methods, attach_arguments, for_argument, \ execute_before -from .decorators import wraps, queue_get, chain_functions, has_keys, short_none +from .decorators import wraps, chain_functions, has_keys, short_none from .preconditions import postcondition, precondition +from .flow_control import loop_while, queue_get __all__ = ['execute_before', 'postcondition', 'precondition', 'wraps', 'queue_get', 'chain_functions', - 'has_keys', 'short_none', 'auto_adapt_to_methods', 'attach_arguments', 'for_argument'] + 'has_keys', 'short_none', 'auto_adapt_to_methods', 'attach_arguments', 'for_argument', + 'loop_while'] diff --git a/satella/coding/decorators/decorators.py b/satella/coding/decorators/decorators.py index 198395a1..7d0e2c54 100644 --- a/satella/coding/decorators/decorators.py +++ b/satella/coding/decorators/decorators.py @@ -9,87 +9,12 @@ U = tp.TypeVar('U') Expression = tp.NewType('Expression', str) ExcType = tp.Type[Exception] + # noinspection PyPep8Naming def _TRUE(x): return True -Queue = tp.TypeVar('Queue') - - -def queue_get(queue_getter: tp.Union[str, tp.Callable[[object], Queue]], timeout: tp.Optional[float] = None, - exception_empty: tp.Union[ExcType, tp.Tuple[ExcType, ...]]=queue.Empty, - queue_get_method: tp.Callable[[Queue, tp.Optional[float]], tp.Any] = - lambda x, timeout: x.get( - timeout=timeout), - method_to_execute_on_empty: tp.Optional[tp.Union[str, tp.Callable]] = None): - """ - A decorator for class methods that consume from a queue. - - Timeout of None means block forever. - - First attribute of the decorator-given function must be a normal instance method - accepting an element taken from the queue, so it must accepts two arguments - first is - self, second is the element from the queue. - - :param queue_getter: a callable that will render us the queue, or a string, which will be - translated to a property name - :param timeout: a timeout to wait. If timeout happens, simple no-op will be done and None - will be returned. - :param exception_empty: exception (or a tuple of exceptions) that are raised on queue being - empty. - :param queue_get_method: a method to invoke on this queue. Accepts two arguments - the first - is the queue, the second is the timeout. It has to follow the type signature given. - :param method_to_execute_on_empty: a callable, or a name of the method to be executed (with no - arguments other than self) to execute in case queue.Empty was raised. Can be a callable - - in that case it should expect no arguments, or can be a string, which will be assumed to be - a method name - - Use instead of: - - >>> class QueueProcessor: - >>> def __init__(self, queue): - >>> self.queue = queue - >>> def do(self): - >>> try: - >>> msg = self.queue.get(timeout=TIMEOUT) - >>> except queue.Empty: - >>> return - - Instead of aforementioned code, please use: - - >>> class QueueProcessor: - >>> def __init__(self, queue): - >>> self.queue = queue - >>> @queue_get(lambda self: self.queue, timeout=TIMEOUT) - >>> def do(self, msg): - >>> ... - """ - if isinstance(queue_getter, str): - my_queue_getter = lambda x: getattr(x, queue_getter) - else: - my_queue_getter = queue_getter - - def outer(fun): - @wraps(fun) - def inner(self): - try: - que = my_queue_getter(self) - item = queue_get_method(que, timeout) - return fun(self, item) - except exception_empty: - if method_to_execute_on_empty is not None: - if callable(method_to_execute_on_empty): - method_to_execute_on_empty() - elif isinstance(method_to_execute_on_empty, str): - method = getattr(self, method_to_execute_on_empty) - method() - - return inner - - return outer - - # taken from https://stackoverflow.com/questions/1288498/using-the-same-decorator-with-arguments-wi\ # th-functions-and-methods def chain_functions(fun_first: tp.Callable[..., tp.Union[tp.Tuple[tp.Tuple, tp.Dict], diff --git a/satella/coding/decorators/flow_control.py b/satella/coding/decorators/flow_control.py new file mode 100644 index 00000000..512d86eb --- /dev/null +++ b/satella/coding/decorators/flow_control.py @@ -0,0 +1,120 @@ +import typing as tp +import queue +from .decorators import wraps, ExcType + + +Queue = tp.TypeVar('Queue') + + +def queue_get(queue_getter: tp.Union[str, tp.Callable[[object], Queue]], timeout: tp.Optional[float] = None, + exception_empty: tp.Union[ExcType, tp.Tuple[ExcType, ...]] = queue.Empty, + queue_get_method: tp.Callable[[Queue, tp.Optional[float]], tp.Any] = + lambda x, timeout: x.get( + timeout=timeout), + method_to_execute_on_empty: tp.Optional[tp.Union[str, tp.Callable]] = None): + """ + A decorator for class methods that consume from a queue. + + Timeout of None means block forever. + + First attribute of the decorator-given function must be a normal instance method + accepting an element taken from the queue, so it must accepts two arguments - first is + self, second is the element from the queue. + + :param queue_getter: a callable that will render us the queue, or a string, which will be + translated to a property name + :param timeout: a timeout to wait. If timeout happens, simple no-op will be done and None + will be returned. + :param exception_empty: exception (or a tuple of exceptions) that are raised on queue being + empty. + :param queue_get_method: a method to invoke on this queue. Accepts two arguments - the first + is the queue, the second is the timeout. It has to follow the type signature given. + :param method_to_execute_on_empty: a callable, or a name of the method to be executed (with no + arguments other than self) to execute in case queue.Empty was raised. Can be a callable - + in that case it should expect no arguments, or can be a string, which will be assumed to be + a method name + + Use instead of: + + >>> class QueueProcessor: + >>> def __init__(self, queue): + >>> self.queue = queue + >>> def do(self): + >>> try: + >>> msg = self.queue.get(timeout=TIMEOUT) + >>> except queue.Empty: + >>> return + + Instead of aforementioned code, please use: + + >>> class QueueProcessor: + >>> def __init__(self, queue): + >>> self.queue = queue + >>> @queue_get(lambda self: self.queue, timeout=TIMEOUT) + >>> def do(self, msg): + >>> ... + """ + if isinstance(queue_getter, str): + my_queue_getter = lambda x: getattr(x, queue_getter) + else: + my_queue_getter = queue_getter + + def outer(fun): + @wraps(fun) + def inner(self): + try: + que = my_queue_getter(self) + item = queue_get_method(que, timeout) + return fun(self, item) + except exception_empty: + if method_to_execute_on_empty is not None: + if callable(method_to_execute_on_empty): + method_to_execute_on_empty() + elif isinstance(method_to_execute_on_empty, str): + method = getattr(self, method_to_execute_on_empty) + method() + + return inner + + return outer + + +def loop_while(pred: tp.Union[tp.Callable[[tp.Any], bool], + tp.Callable[[], bool]] = lambda: True): + """ + Decorator to loop the following function while predicate called on it's first argument is True. + + Use to mostly loop class methods basing off classes, like: + + >>> class Terminable: + >>> terminated = False + >>> @loop_while(x.terminated == False) + >>> def run(self): + >>> ... + + You can also loop standard functions, like this: + + >>> a = {'terminating': False} + >>> @loop_while(lambda: not a['terminating']) + >>> def execute_while(): + >>> ... + + :param pred: predicate to evaluate. Can accept either one element, in this case + it will be fed the class instance, or accept no arguments, in which case + it will be considered to annotate a method + """ + def outer(fun): + @wraps(fun) + def inner(*args, **kwargs): + if len(args) > 0: + p = pred(args[0]) + else: + p = pred() + while p: + fun(*args, **kwargs) + if len(args) > 0: + p = pred(args[0]) + else: + p = pred() + return inner + return outer diff --git a/tests/test_coding/test_decorators.py b/tests/test_coding/test_decorators.py index 65311c58..29d4c75f 100644 --- a/tests/test_coding/test_decorators.py +++ b/tests/test_coding/test_decorators.py @@ -6,7 +6,8 @@ from socket import socket from satella.coding import wraps, chain_functions, postcondition, \ log_exceptions, queue_get, precondition, short_none from satella.coding.decorators import auto_adapt_to_methods, attach_arguments, \ - execute_before + execute_before, loop_while +from satella.coding.predicates import x from satella.exceptions import PreconditionError logger = logging.getLogger(__name__) @@ -14,6 +15,28 @@ logger = logging.getLogger(__name__) class TestDecorators(unittest.TestCase): + def test_loop_while(self): + class MyLooped: + terminating = False + i = 0 + + @loop_while(x.i < 10) + def run(self): + self.i += 1 + + a = MyLooped() + a.run() + self.assertGreaterEqual(a.i, 10) + b = {'i': 0} + + @loop_while(lambda: b['i'] < 10) + def run(): + nonlocal b + b['i'] += 1 + + run() + self.assertGreaterEqual(b['i'], 10) + def test_execute_before(self): a = 0 -- GitLab