Decorators

satella.coding.decorators.execute_if_attribute_none(attribute: str)

Decorator for instancemethods.

This will execute the function only if provided attribute is None. Otherwise it will return a None

Parameters:

attribute – name of the attribute to check

satella.coding.decorators.execute_if_attribute_not_none(attribute: str)

Decorator for instancemethods.

This will execute the function only if provided attribute is not None. Otherwise it will return a None

Parameters:

attribute – name of the attribute to check

satella.coding.decorators.transform_result(expr: str)

A decorator transforming the result value of a function by a Python expression.

The result is feeded as the local variable “x”, while arguments are fed as if they were expressed as arguments, eg:

>>> @transform_result('x*a')
>>> def square(a):
>>>     return a

Parameters:

expr – Python string expression

satella.coding.decorators.transform_arguments(**expressions: str)

A decorator transforming the arguments of a function prior to it’s execution.

The arguments are always bound as if they were available in the function.

The expressions always operate on “old” arguments

>>> @transform_arguments(a='a*a')
>>> def square(a):
>>>     return a

Parameters:

expressions – Python strings that are meant to be evaluated

satella.coding.decorators.retry(times: int | None = None, exc_classes: ~typing.Type[Exception] | ~typing.Tuple[~typing.Type[Exception], ...] = <class 'Exception'>, on_failure: ~typing.Callable[[Exception], None] = <function <lambda>>, swallow_exception: bool = True, call_on_failure: ~typing.Callable[[Exception], None] | None = None, call_on_success: ~typing.Callable[[int], None] | None = None, call_between_retries: ~typing.Callable[[Exception], None] | None = None)

A decorator retrying given operation, failing it when an exception shows up.

Essentially this:

>>> @retry(3, RuntimeError)
>>> def do_op():
>>>     .. do op ..

Is a syntactic sugar for this:

>>> for i in range(3):
>>>     try:
>>>         .. do op ..
>>>         break
>>>     except RuntimeError:
>>>         pass

Warning

Retry by default will swallow the resulting exception. To avoid this behaviour set the swallow_exception parameter to False.

Parameters:

• times – maximum times to retry this operation. By default (None) retry until the world ends.

• exc_classes – exception classes that should be considered failure. By default. catches everything (Exception).

• on_failure – callable to call when it fails times times. Callable will be called with a single argument, exception instance that was raised last. That exception will be swallowed, unless swallow_exception is set to False

• swallow_exception – the last exception will be swallowed, unless this is set to False

• call_on_failure – a callable that will be called upon failing to do this, with an exception as it’s sole argument. It’s result will be discarded. Deprecated, use on_failure instead.

• call_on_success – a callable that will be called with a single argument: the number of retries that it took to finish the job. It’s result will be discarded.

• call_between_retries – called between retries with a single argument, the Exception instance that forced the retry.

Returns:

function result

satella.coding.decorators.memoize(fun)

A thread safe memoizer based on function’s ONLY positional arguments.

Note that this will make your function execute it at most one thread, the remaining ones will have to wait.

Usage example:

>>> @memoize
>>> def expensive_but_idempotent_operation(a):
>>>     ...

>>> a = expensive_but_idempotent_operation(2)
>>> b = expensive_but_idempotent_operation(2)   # is much faster than computing the value anew

satella.coding.decorators.call_method_on_exception(exc_classes, method_name, *args, **kwargs)

A decorator for instance methods to call provided method with given arguments if given call fails.

Example use:

>>> class Test:
>>>     def close(self):
>>>         ...
>>>     @call_method_on_exception(ValueError, 'close')
>>>     def read(self, bytes):
>>>         raise ValueError()

Exception class determination is done by isinstance, so you can go wild with metaclassing. Exception will be swallowed. The return value will be taken from the called function.

Note that the called method must be an instance method.

Parameters:

• exc_classes – a tuple or an instance class to which react

• method_name – name of the method. It must be gettable by getattr

• args – arguments to pass to given method

• kwargs – keyword arguments to pass to given method

satella.coding.decorators.cache_memoize(cache_duration: float, time_getter: ~typing.Callable[[], float] = <built-in function monotonic>)

A thread-safe memoizer that memoizes the return value for at most cache_duration seconds.

Parameters:

• cache_duration – cache validity, in seconds

• time_getter – a callable without arguments that yields us a time marker

Usage example:

>>> @cache_memoize(10)
>>> def expensive_but_idempotent_operation(a):
>>>     ...

>>> a = expensive_but_idempotent_operation(2)
>>> b = expensive_but_idempotent_operation(2)   # is much faster than computing the value anew
>>> time.sleep(10)
>>> c = expensive_but_idempotent_operation(2)   # function body is computed anew

satella.coding.decorators.queue_get(queue_getter: str | ~typing.Callable[[object], ~satella.coding.decorators.flow_control.Queue], timeout: float | None = None, exception_empty: ~typing.Type[Exception] | ~typing.Tuple[~typing.Type[Exception], ...] = <class '_queue.Empty'>, queue_get_method: ~typing.Callable[[~satella.coding.decorators.flow_control.Queue, float | None], ~typing.Any] = <function <lambda>>, method_to_execute_on_empty: str | ~typing.Callable | None = 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.

Parameters:

• queue_getter – a callable that will render us the queue, or a string, which will be translated to a property name

• timeout – a timeout to wait. If timeout happens, simple no-op will be done and None will be returned.

• exception_empty – exception (or a tuple of exceptions) that are raised on queue being empty.

• 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.

• 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):
>>>         ...

satella.coding.decorators.copy_arguments(deep_copy: bool = False) → Callable

Make every argument passe to this function be copied.

This way you can pass dictionaries to the function that would normally have modified them.

Use like this:

>>> @copy_arguments()
>>> def alter_dict(dct: dict)
>>>     return dct.pop('a')

Now you can use it like this:

>>> b = {'a': 5}
>>> assert alter_dict(b) == 5
>>> assert b == {'a': 5}

Parameters:

deep_copy – whether to use deepcopy instead of a plain copy

satella.coding.decorators.loop_while(pred: ~typing.Callable[[~satella.coding.typing.T], bool] | ~typing.Callable[[], bool] = <function <lambda>>)

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:

>>> from satella.coding.predicates import x
>>> 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():
>>>     ...

Parameters:

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.

Note that the function you decorate may only take arguments if it’s a class method. If it’s a standard method, then it should not take any arguments.

satella.coding.decorators.repeat_forever(fun)

A decorator that will place your function inside a while True loop.

satella.coding.for_argument(*t_ops: Callable[[T], U] | str, **t_kwops: Callable[[T], U] | str)

Calls a callable for each of the arguments. Pass None if you do not wish to process given argument.

returns is a special keyword, a callable to process the result through

Use like:

>>> @for_argument(int, str, typed=bool, returns=int)
>>> def check(val1, val2, typed='True'):
>>>     if typed:
>>>         return val1 + int(val2)

for_argument can also accept strings as expressions:

>>> @for_argument('x*2')
>>> def accept_two(x):
>>>     assert x == 2
>>> accept_two(1)

for_argument will also recognize default values:

>>> @for_argument(k=int)
>>> def for_arg(k='5')
>>>     print(repr(k))
>>> for_arg()
will print `5` instead of `'5'`.

Note that for_argument is quite slow when it comes to having default values in the function signature. Best to avoid it if you need speed.

If it detects that the function that you passed does not use default values, it will use the faster implementation.

satella.coding.chain_functions(fun_first: Callable[[...], Tuple[Tuple, Dict] | Dict | Tuple]) → Callable

A decorator to chain function calls. This function is expected to return:

• a 2-tuple [tp.Tuple, tp.Dict] - args and kwargs for the next function

• tp.Dict - only kwargs will be passed

• any other tuple - only args will be passed

• any other type - will be passed as a first argument

of arguments to pass to wrapped function. So this

>>> def test3(...):
>>>     ...
>>> def test2(...):
>>>     ...
>>> def test(...):
>>>     args, kwargs = test2(...)
>>>     return test(3)
>>> v = test(a, b, c)

Is equivalent to this: >>> @chain_functions >>> def test2(…): >>> … >>> @test2 >>> def test3(…): >>> … >>> v = test3(a, b, c)

satella.coding.auto_adapt_to_methods(decorator)

Allows you to use the same decorator on methods and functions, hiding the self argument from the decorator.

Usage:

>>> @auto_adapt_to_methods
>>> def times_two(fun):
>>>     def outer(a):
>>>         return fun(a*2)
>>>     return outer
>>> class Test:
>>>     @times_two
>>>     def twice(self, a):
>>>         return a*2
>>> @times_two
>>> def twice(a):
>>>     return a*2
>>> assert Test().twice(2) == 4
>>> assert twice(2) == 4

satella.coding.attach_arguments(*args, **kwargs)

Return a decorator that passes extra arguments to the function.

Example:

>>> @attach_arguments(2, label='value')
>>> def print_args(*args, **kwargs):
>>>     print(args, kwargs)
>>> print_args(3, 4, key='value')

will print

>>> (3, 4, 2) {'key': 'value', 'label': 'value'}

Arguments given in attach_arguments will take precedence in case of key collisions.

A functools.wraps() equivalent, but for classes

satella.coding.wraps(cls_to_wrap: Type) → Callable[[Type], Type]

A functools.wraps() but for classes.

As a matter of fact, this can replace functools.wraps() entirely. This replaces __doc__, __name__, __module__ and __annotations__. It also sets a correct __wrapped__.

Parameters:

cls_to_wrap – class to wrap

satella.coding.decorators.execute_before(callable_: Callable) → Callable

Wrapper to create wrappers which execute callable before function launch.

Use like this:

>>> @execute_before
>>> def do_things():
>>>     print('Things are done')

Then the following will print Things are done:

>>> @do_things
>>> def nothing():
>>>     ...
>>> nothing()

You can even specify custom parameters for the callable:

>>> @execute_before
>>> def i_am_2(two):
>>>     assert two == 2
>>> @i_am_2(2)
>>> def run_me():
>>>     pass

satella.coding.decorators.return_as_list(ignore_nulls: bool = False)

Enables you to write a list-returning functions using a decorator. Example:

>>> def make_a_list(lst):
>>>     output = []
>>>     for item in lst:
>>>         output.append(item)
>>>     return output

Is equivalent to:

>>> @return_as_list()
>>> def make_a_list(lst):
>>>     for item in lst:
>>>         yield item

Essentially a syntactic sugar for @for_argument(returns=list)

Parameters:

ignore_nulls – if True, then if your function yields None, it won’t be appended.

satella.coding.decorators.default_return(v: Any)

Makes the decorated function return v instead of None, if it would return None. If it would return something else, that else is returned.

Eg:

>>> @default_return(5)
>>> def return(v):
>>>     return v
>>> assert return(None) == 5
>>> assert return(2) == 2

Parameters:

v – value to return if calling the function would return None

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:

satella.coding.precondition(*t_ops: Callable[[T], bool] | Expression, **kw_opts: Callable[[T], bool] | Expression)

Check that a precondition happens for given parameter.

You can do it like this:

>>> @precondition(lambda x: x == 1)
>>> def return_two(x):
>>>     return x*2

or

>>> @precondition('x == 1')
>>> def return_two(x):
>>>     ..

You can use all standard locals in precondition.

You function call will return a PreconditionError (subclass of ValueError) if a precondition fails.

A precondition of None will always be true.

Keyword arguments are supported as well. Note that precondition for them will be checked only if they are passed, so make your default arguments obey the precondition, because it won’t be checked if the default value is used.

And here are some helper functions for it:

has_keys asserts that a dictionary has all the keys necessary.

satella.coding.has_keys(keys: List[str])

A decorator for asserting that a dictionary has given keys. Will raise PreconditionError if it doesn’t.

This outputs a callable that accepts a dict and returns True if it has all the keys necessary.

Returns True if the dict has all necessary keys.

This is meant to be used in conjunction with @precondition

Deprecated since version 2.14.22.

Parameters:

keys – list of keys to expect

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.

satella.coding.short_none(clb: Expression | Callable[[T], U]) → Callable[[T | None], U | None]

Accept a callable. Return a callable that executes it only if passed a no-None arg, and returns its result. If passed a None, return a None

callable can also be a string, in this case it will be appended to lambda x: and eval’d

Parameters:

clb – callable/1->1

Returns:

a modified callable

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

satella.coding.postcondition(condition: Callable[[T], bool] | Expression)

Return a decorator, asserting that result of this function, called with provided callable, is True, else the function will raise PreconditionError.

Parameters:

condition – callable that accepts a single argument, the return value of the function. Can be also a string, in which case it is an expression about the value x of return