satella.coding.decorators package¶
Submodules¶
satella.coding.decorators.arguments module¶
- satella.coding.decorators.arguments.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.
- satella.coding.decorators.arguments.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.decorators.arguments.cached_property(prop_name, assume_not_loaded=None)¶
A decorator to use to create cached properties.
You job is to only write the value returner. If the value is currently assume_not_loaded (None by default) your property method will be called. Otherwise it will be served from cached attribute, whose value you provide as parameter.
Use as follows:
>>> class Example: >>> def __init__(self): >>> self._a = None >>> @property >>> @cached_property('_a') >>> def a(self) -> str: >>> return 'abc' >>> a = Example() >>> assert a.a == 'abc' >>> assert a._a == 'abc'
- Parameters:
prop_name (str) – Name of property to store the value in
assume_not_loaded – Value if currently the attribute is equal to this, it is assumed to not have been loaded
- satella.coding.decorators.arguments.copy_arguments(deep_copy=False)¶
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 (bool) – whether to use deepcopy instead of a plain copy
- Return type:
Callable
- satella.coding.decorators.arguments.execute_before(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
- Parameters:
callable_ (Callable) –
- Return type:
Callable
- satella.coding.decorators.arguments.execute_if_attribute_none(attribute)¶
Decorator for instancemethods.
This will execute the function only if provided attribute is None. Otherwise it will return a None
- Parameters:
attribute (str) – name of the attribute to check
- satella.coding.decorators.arguments.execute_if_attribute_not_none(attribute)¶
Decorator for instancemethods.
This will execute the function only if provided attribute is not None. Otherwise it will return a None
- Parameters:
attribute (str) – name of the attribute to check
- satella.coding.decorators.arguments.for_argument(*t_ops, **t_kwops)¶
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.
- Parameters:
t_ops (Union[Callable[[T], U], str]) –
t_kwops (Union[Callable[[T], U], str]) –
- satella.coding.decorators.arguments.replace_argument_if(arg_name, structure, instance_of=None, predicate=None)¶
Examine arguments of the callable that will be decorated with this.
If argument arg_name is found to be an instance of instance_of, it will be replaced by a structure defined a structure.
- Parameters:
arg_name (str) – argument to replace
instance_of (Optional[Union[Type, Tuple[Type, ...]]]) – type
predicate (Optional[Callable[[T], bool]]) – alternative condition of replacement. If this is given, predicate is called on the value of the argument and replacement is done if it returns True
structure (Union[dict, list, tuple, PredicateClass]) – a callable that takes original argument and returns new, or a structure made of these
- satella.coding.decorators.arguments.transform_arguments(**expressions)¶
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 (str) – Python strings that are meant to be evaluated
- satella.coding.decorators.arguments.transform_result(expr)¶
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 (str) – Python string expression
satella.coding.decorators.decorators module¶
- satella.coding.decorators.decorators.cache_memoize(cache_duration, time_getter=<built-in function monotonic>)¶
A thread-safe memoizer that memoizes the return value for at most cache_duration seconds.
- Parameters:
cache_duration (float) – cache validity, in seconds
time_getter (Callable[[], float]) – 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.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.decorators.chain_functions(fun_first)¶
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)
- Parameters:
fun_first (Callable[[...], Union[Tuple[Tuple, Dict], Dict, Tuple]]) –
- Return type:
Callable
- satella.coding.decorators.decorators.default_return(v)¶
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 (Any) – value to return if calling the function would return None
- satella.coding.decorators.decorators.has_keys(keys)¶
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[str]) – list of keys to expect
- satella.coding.decorators.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.decorators.return_as_list(ignore_nulls=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 (bool) – if True, then if your function yields None, it won’t be appended.
- satella.coding.decorators.decorators.short_none(clb)¶
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 (Union[Expression, Callable[[T], U]]) – callable/1->1
- Returns:
a modified callable
- Return type:
Callable[[Optional[T]], Optional[U]]
- satella.coding.decorators.decorators.wraps(cls_to_wrap)¶
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 (Type) – class to wrap
- Return type:
Callable[[Type], Type]
satella.coding.decorators.flow_control module¶
- satella.coding.decorators.flow_control.loop_while(pred=<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 (Union[Callable[[T], bool], Callable[[], bool]]) – 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.flow_control.queue_get(queue_getter, timeout=None, exception_empty=<class '_queue.Empty'>, queue_get_method=<function <lambda>>, method_to_execute_on_empty=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 (Union[str, Callable[[object], Queue]]) – a callable that will render us the queue, or a string, which will be translated to a property name
timeout (Optional[float]) – a timeout to wait. If timeout happens, simple no-op will be done and None will be returned.
exception_empty (Union[Type[Exception], Tuple[Type[Exception], ...]]) – exception (or a tuple of exceptions) that are raised on queue being empty.
queue_get_method (Callable[[Queue, Optional[float]], Any]) – 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 (Optional[Union[str, Callable]]) – 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.flow_control.repeat_forever(fun)¶
A decorator that will place your function inside a while True loop.
satella.coding.decorators.preconditions module¶
- satella.coding.decorators.preconditions.postcondition(condition)¶
Return a decorator, asserting that result of this function, called with provided callable, is True, else the function will raise PreconditionError.
- Parameters:
condition (Union[Callable[[T], bool], Expression]) – 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
- satella.coding.decorators.preconditions.precondition(*t_ops, **kw_opts)¶
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.
- Parameters:
t_ops (Union[Callable[[T], bool], Expression]) –
kw_opts (Union[Callable[[T], bool], Expression]) –
satella.coding.decorators.retry_dec module¶
- satella.coding.decorators.retry_dec.retry(times=None, exc_classes=<class 'Exception'>, on_failure=<function <lambda>>, swallow_exception=True, call_on_failure=None, call_on_success=None, call_between_retries=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 (Optional[int]) – maximum times to retry this operation. By default (None) retry until the world ends.
exc_classes (Union[Type[Exception], Tuple[Type[Exception], ...]]) – exception classes that should be considered failure. By default. catches everything (Exception).
on_failure (Callable[[Exception], None]) – 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 (bool) – the last exception will be swallowed, unless this is set to False
call_on_failure (Optional[Callable[[Exception], None]]) – 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 (Optional[Callable[[int], None]]) – 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 (Optional[Callable[[Exception], None]]) – called between retries with a single argument, the Exception instance that forced the retry.
- Returns:
function result
Module contents¶
- satella.coding.decorators.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.
- satella.coding.decorators.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.decorators.cache_memoize(cache_duration, time_getter=<built-in function monotonic>)¶
A thread-safe memoizer that memoizes the return value for at most cache_duration seconds.
- Parameters:
cache_duration (float) – cache validity, in seconds
time_getter (Callable[[], float]) – 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.cached_property(prop_name, assume_not_loaded=None)¶
A decorator to use to create cached properties.
You job is to only write the value returner. If the value is currently assume_not_loaded (None by default) your property method will be called. Otherwise it will be served from cached attribute, whose value you provide as parameter.
Use as follows:
>>> class Example: >>> def __init__(self): >>> self._a = None >>> @property >>> @cached_property('_a') >>> def a(self) -> str: >>> return 'abc' >>> a = Example() >>> assert a.a == 'abc' >>> assert a._a == 'abc'
- Parameters:
prop_name (str) – Name of property to store the value in
assume_not_loaded – Value if currently the attribute is equal to this, it is assumed to not have been loaded
- 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.chain_functions(fun_first)¶
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)
- Parameters:
fun_first (Callable[[...], Union[Tuple[Tuple, Dict], Dict, Tuple]]) –
- Return type:
Callable
- satella.coding.decorators.copy_arguments(deep_copy=False)¶
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 (bool) – whether to use deepcopy instead of a plain copy
- Return type:
Callable
- satella.coding.decorators.default_return(v)¶
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 (Any) – value to return if calling the function would return None
- satella.coding.decorators.execute_before(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
- Parameters:
callable_ (Callable) –
- Return type:
Callable
- satella.coding.decorators.execute_if_attribute_none(attribute)¶
Decorator for instancemethods.
This will execute the function only if provided attribute is None. Otherwise it will return a None
- Parameters:
attribute (str) – name of the attribute to check
- satella.coding.decorators.execute_if_attribute_not_none(attribute)¶
Decorator for instancemethods.
This will execute the function only if provided attribute is not None. Otherwise it will return a None
- Parameters:
attribute (str) – name of the attribute to check
- satella.coding.decorators.for_argument(*t_ops, **t_kwops)¶
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.
- Parameters:
t_ops (Union[Callable[[T], U], str]) –
t_kwops (Union[Callable[[T], U], str]) –
- satella.coding.decorators.has_keys(keys)¶
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[str]) – list of keys to expect
- satella.coding.decorators.loop_while(pred=<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 (Union[Callable[[T], bool], Callable[[], bool]]) – 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.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.postcondition(condition)¶
Return a decorator, asserting that result of this function, called with provided callable, is True, else the function will raise PreconditionError.
- Parameters:
condition (Union[Callable[[T], bool], Expression]) – 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
- satella.coding.decorators.precondition(*t_ops, **kw_opts)¶
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.
- Parameters:
t_ops (Union[Callable[[T], bool], Expression]) –
kw_opts (Union[Callable[[T], bool], Expression]) –
- satella.coding.decorators.queue_get(queue_getter, timeout=None, exception_empty=<class '_queue.Empty'>, queue_get_method=<function <lambda>>, method_to_execute_on_empty=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 (Union[str, Callable[[object], Queue]]) – a callable that will render us the queue, or a string, which will be translated to a property name
timeout (Optional[float]) – a timeout to wait. If timeout happens, simple no-op will be done and None will be returned.
exception_empty (Union[Type[Exception], Tuple[Type[Exception], ...]]) – exception (or a tuple of exceptions) that are raised on queue being empty.
queue_get_method (Callable[[Queue, Optional[float]], Any]) – 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 (Optional[Union[str, Callable]]) – 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.repeat_forever(fun)¶
A decorator that will place your function inside a while True loop.
- satella.coding.decorators.replace_argument_if(arg_name, structure, instance_of=None, predicate=None)¶
Examine arguments of the callable that will be decorated with this.
If argument arg_name is found to be an instance of instance_of, it will be replaced by a structure defined a structure.
- Parameters:
arg_name (str) – argument to replace
instance_of (Optional[Union[Type, Tuple[Type, ...]]]) – type
predicate (Optional[Callable[[T], bool]]) – alternative condition of replacement. If this is given, predicate is called on the value of the argument and replacement is done if it returns True
structure (Union[dict, list, tuple, PredicateClass]) – a callable that takes original argument and returns new, or a structure made of these
- satella.coding.decorators.retry(times=None, exc_classes=<class 'Exception'>, on_failure=<function <lambda>>, swallow_exception=True, call_on_failure=None, call_on_success=None, call_between_retries=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 (Optional[int]) – maximum times to retry this operation. By default (None) retry until the world ends.
exc_classes (Union[Type[Exception], Tuple[Type[Exception], ...]]) – exception classes that should be considered failure. By default. catches everything (Exception).
on_failure (Callable[[Exception], None]) – 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 (bool) – the last exception will be swallowed, unless this is set to False
call_on_failure (Optional[Callable[[Exception], None]]) – 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 (Optional[Callable[[int], None]]) – 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 (Optional[Callable[[Exception], None]]) – called between retries with a single argument, the Exception instance that forced the retry.
- Returns:
function result
- satella.coding.decorators.return_as_list(ignore_nulls=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 (bool) – if True, then if your function yields None, it won’t be appended.
- satella.coding.decorators.short_none(clb)¶
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 (Union[Expression, Callable[[T], U]]) – callable/1->1
- Returns:
a modified callable
- Return type:
Callable[[Optional[T]], Optional[U]]
- satella.coding.decorators.transform_arguments(**expressions)¶
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 (str) – Python strings that are meant to be evaluated
- satella.coding.decorators.transform_result(expr)¶
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 (str) – Python string expression
- satella.coding.decorators.wraps(cls_to_wrap)¶
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 (Type) – class to wrap
- Return type:
Callable[[Type], Type]