satella.coding package¶
Subpackages¶
- satella.coding.concurrent package
- Subpackages
- Submodules
- satella.coding.concurrent.atomic module
- satella.coding.concurrent.callablegroup module
- satella.coding.concurrent.functions module
- satella.coding.concurrent.id_allocator module
- satella.coding.concurrent.list_processor module
- satella.coding.concurrent.locked_dataset module
- satella.coding.concurrent.locked_structure module
- satella.coding.concurrent.monitor module
- satella.coding.concurrent.queue module
- satella.coding.concurrent.sync module
- satella.coding.concurrent.thread module
- satella.coding.concurrent.thread_collection module
- satella.coding.concurrent.timer module
- satella.coding.concurrent.value module
- Module contents
- satella.coding.decorators package
- satella.coding.resources package
- satella.coding.sequences package
- satella.coding.structures package
- Subpackages
- satella.coding.structures.dictionaries package
- Submodules
- satella.coding.structures.dictionaries.cache_dict module
- satella.coding.structures.dictionaries.counting module
- satella.coding.structures.dictionaries.default module
- satella.coding.structures.dictionaries.dict_object module
- satella.coding.structures.dictionaries.expiring module
- satella.coding.structures.dictionaries.objects module
- satella.coding.structures.dictionaries.writeback_cache module
- Module contents
- satella.coding.structures.heaps package
- satella.coding.structures.mixins package
- satella.coding.structures.dictionaries package
- Submodules
- satella.coding.structures.hashable_objects module
- satella.coding.structures.immutable module
- satella.coding.structures.lru module
- satella.coding.structures.proxy module
- satella.coding.structures.push_iterable module
- satella.coding.structures.queues module
- satella.coding.structures.ranking module
- satella.coding.structures.singleton module
- satella.coding.structures.sorted_list module
- satella.coding.structures.sparse_matrix module
- satella.coding.structures.syncable_droppable module
- satella.coding.structures.tuples module
- satella.coding.structures.typednamedtuple module
- satella.coding.structures.zip_dict module
- Module contents
- Subpackages
- satella.coding.transforms package
- Submodules
- satella.coding.transforms.base64 module
- satella.coding.transforms.interpol module
- satella.coding.transforms.jsonify module
- satella.coding.transforms.merge_list module
- satella.coding.transforms.merger module
- satella.coding.transforms.misc module
- satella.coding.transforms.percentile module
- satella.coding.transforms.predicates module
- satella.coding.transforms.words module
- Module contents
Submodules¶
satella.coding.algos module¶
- satella.coding.algos.merge_dicts(v1, v2)¶
Try to merge two dicts/list together. If key collision is found, value from v2 will be taken.
If the objects aren’t dicts or lists, v2 will be returned.
Lists will be concatenated, and dicts updated. v1 will be updated in-place!
- Parameters:
v1 (Any) –
v2 (Any) –
- Return type:
Any
satella.coding.ctxt_managers module¶
- class satella.coding.ctxt_managers.EmptyContextManager(*args, **kwargs)¶
Bases:
object
A context manager that does nothing. Only to support conditional change of context managers, eg in such a way:
>>> if tx_enabled: >>> ctxt = transaction.atomic >>> else: >>> ctxt = EmptyContextManager() >>> with ctxt: >>> ...
Note that it will accept any parameters, and then throw them on the ground.
- satella.coding.ctxt_managers.wrap_callable_in_context_manager(clbl, ctxt_mgr, *my_args, **my_kwargs)¶
Wrap a callable in context manager.
Roughly equivalent to:
>>> def inner(*args, **kwargs): >>> with ctxt_mgr(*my_args, **my_kwargs): >>> return clbl(*args, **kwargs) >>> return inner
To be used as:
>>> clbl = wrap_callable_in_context_manager(lambda y: 5, tracing.start_new_span, 'New span')
satella.coding.deep_compare module¶
- exception satella.coding.deep_compare.Inequal(obj1, obj2, reason)¶
Bases:
Exception
An exception raised by
deep_compare()
if two objects don’t match- Variables:
obj1 – first object that was not equal, or key name
obj2 – second object that was not equal, or None
reason – (
InequalityReason
) reason for inequality
- Parameters:
reason (InequalityReason) –
- class satella.coding.deep_compare.InequalityReason(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)¶
Bases:
IntEnum
- KEY_NOT_FOUND = 2¶
key given as obj1 was not found
- LENGTH_MISMATCH = 1¶
length didn’t match
- NOT_EQUAL = 0¶
direct eq yielded not equal
- satella.coding.deep_compare.assert_equal(a, b)¶
Assert that two values are equal. If not, an
satella.coding.Inequality
exception will be thrown.Objects are tried to compare using it’s
__eq__
.- Parameters:
a – first value to compare
b – second value to compare
- Raises:
Inequal – objects were not equal
satella.coding.deleters module¶
- class satella.coding.deleters.DictDeleter(dict_to_process)¶
Bases:
object
Having problems deleting entries from your dict while iterating on them? No problem. Just swap the following:
>>> keys_to_delete = [] >>> for key, value in my_dict.items(): >>> if value.should_delete(): >>> keys_to_delete.append(key) >>> for key in keys_to_delete: >>> del my_dict[key]
With the following:
>>> with DictDeleter(my_list) as ld: >>> for key, value in ld.items(): >>> if value.should_delete(): >>> ld.delete()
Note that a single DictDeleter running from a single context must be iterated on by only a single Thread as it keeps the state of iterator in itself, to prevent allocating new objects and slowing things down too much.
This allocates only a single object per a call to delete().
- Parameters:
dict_to_process (MutableMapping) –
- current_iterator¶
- current_key¶
- delete()¶
- Return type:
None
- dict_to_process¶
- items()¶
- Return type:
- iter_mode¶
- keys()¶
- Return type:
- keys_to_delete¶
- values()¶
- Return type:
- class satella.coding.deleters.ListDeleter(list_to_process, direction=0)¶
Bases:
Generic
[T
]Having problems deleting entries from your list while iterating on them? No problem. Just swap the following:
>>> entries_to_delete = [] >>> for entry in my_list: >>> if entry.should_delete(): >>> entries_to_delete.append(entry) >>> for entry in entries_to_delete: >>> my_list.remove(entry)
With the following:
>>> with ListDeleter(my_list) as ld: >>> for entry in ld: >>> if entry.should_delete(): >>> ld.delete()
You can also use the alternative syntax of: >>> ld = ListDeleter(my_list) >>> while True: >>> try: >>> v = ld.next() >>> except StopIteration: >>> break >>> if condition(v): >>> ld.delete() >>> ld.remove_items()
Note that a single ListDeleter running from a single context must be iterated on by only a single Thread as it keeps the state of iterator in itself, to prevent allocating new objects and slowing things down too much.
Note that calling reversed() on this will reset the pointer to the end of the list or the beginning of the list, respectively.
This allocates only a single object per a call to delete().
Calling the list deleter during iteration will yield the element.
You can pass any type of object here, as long as it supports pop(position) and __getitem__
- Parameters:
list_to_process (MutableSequence[T]) –
direction (int) –
- current_index¶
- delete()¶
- Return type:
None
- direction¶
- indices_to_delete: Set[int]¶
- list_to_process¶
- next()¶
- Returns:
the next element
- Raises:
StopIteration – no more entries
- Return type:
T
- prev()¶
Move to previous element, as per ordering.
- Returns:
the previous element
- Raises:
StopIteration – list is already at the first element!
- Return type:
T
- remove_items()¶
After all of the items have been marked for deletion, delete them
- Return type:
None
- removed¶
- property value: T¶
satella.coding.environment module¶
- class satella.coding.environment.Context(parent=None, **variables)¶
Bases:
object
New layer of environment. Can have it’s own variables, or can hoist them onto the parent.
Warning
This is considered experimental. I just haven’t found out a good use case for it yet.
- Parameters:
parent (tp.Optional[Context]) –
- does_exist(val)¶
Does a given value exist on stack for this call of function?
- Parameters:
val (str) –
- Return type:
bool
- push_up(item, value=<object object>)¶
Advance current variable to the top of the card stack.
- Parameters:
item (str) – variable name
value – if not given, current value of given variable will be taken
- Return type:
None
satella.coding.expect_exception module¶
- class satella.coding.expect_exception.expect_exception(exc_to_except, else_raise, *args, **kwargs)¶
Bases:
object
A context manager to use as following:
>>> a = {'test': 2} >>> with expect_exception(KeyError, ValueError, 'KeyError not raised'): >>> a['test2']
If other exception than the expected is raised, it is passed through
- Parameters:
exc_to_except (Union[Type[Exception], Tuple[Type[Exception], ...]]) – a list of exceptions or a single exception to expect
else_raise (Type[Exception]) – raise a particular exception if no exception is raised. This should be a callable that accepts provided args and kwargs and returns an exception instance.
args – args to provide to constructor
kwargs – kwargs to provide to constructor
- else_raise¶
- else_raise_args¶
- else_raise_kwargs¶
- exc_to_except¶
satella.coding.fun_static module¶
- satella.coding.fun_static.static_var(var_name, starting_value=None)¶
Declare a static variable for given function
Use it like:
>>> @static_var('counter', 2) >>> def count(): >>> count.counter += 1
or:
>>> class MyClass: >>> @static_var('counter', 2) >>> def count(self): >>> MyClass.count.counter += 1
- Parameters:
var_name (str) –
starting_value (Optional[Any]) –
satella.coding.generators module¶
- class satella.coding.generators.RunActionAfterGeneratorCompletes(generator, *args, call_despite_closed=False, **kwargs)¶
Bases:
Generator
Run an action after a generator completes. An abstract class.
Please note that this routine will be called only when the generator completes. If you abort it prematurely, via close()
- Parameters:
generator (Generator) – generator to watch for
args – arguments to invoke action_to_run with
call_despite_closed (bool) –
action_to_run()
will be called even if the generator is closedkwargs – keyword arguments to invoke action_to_run with
- abstract action_to_run(*args, **kwargs)¶
This will run when this generator completes. Override it.
- args¶
- call_despite_closed¶
- call_on_exception(exc)¶
This will run when this generator throws any exception inside it’s __next__() or send(). You can reraise it (which is the default behavior if you do not override this).
- Parameters:
exc (Exception) –
- close()¶
Close this generator. Note that this will cause
action_to_run()
not to run
- closed¶
- generator¶
- kwargs¶
- next()¶
- send(value)¶
Send a value to the generator
- throw(_RunActionAfterGeneratorCompletes__typ, _RunActionAfterGeneratorCompletes__val=None, _RunActionAfterGeneratorCompletes__tb=None)¶
Raise an exception in the generator. Return next yielded value or raise StopIteration.
- satella.coding.generators.run_when_generator_completes(gen, call_on_done, *args, **kwargs)¶
Return the generator with call_on_done to be called on when it finishes
- Parameters:
gen (Generator) – generator
call_on_done (Callable) – callable/0 to call on generator’s completion
args – args to pass to the callable
kwargs – kwargs to pass to the callable
- Returns:
generator
- Return type:
satella.coding.iterators module¶
- class satella.coding.iterators.SelfClosingGenerator(generator)¶
Bases:
object
A wrapper to exhaust the generator in response to closing it.
This will allow generators to complete that don’t provide a .close() method.
This will additionally exhaust the generator upon deallocation of the generator.
You can feed it with either generators, or generator-functions, it will behave correctly each time.
You can also use it as a context manager, to decouple finalizing the generator from the GC collection
- Parameters:
generator (Union[Generator, Callable[[Any], Generator]]) –
- close()¶
- Return type:
None
- generator¶
- send(obj)¶
- Parameters:
obj (Any) –
- Return type:
None
- stopped: bool¶
- satella.coding.iterators.chain(*args)¶
Construct an iterator out of provided elements.
If an element is an iterator, or an iterable it will be yielded-from. If it’s not, it will just be yielded.
A cast to iter() is used to determine iteratorness
- Return type:
Iterator
- satella.coding.iterators.exhaust(iterator)¶
Iterate till the end of the iterator, discarding values as they go
- Parameters:
iterator (Iterator) – iterator to exhaust
- Return type:
None
- class satella.coding.iterators.hint_with_length(generator, length, length_factory=None)¶
Bases:
object
Accepting a generator, return it additionally providing a specified __length_hint__
You can provide generator-generating functions as well
- Parameters:
generator (Generator) – generator to decorate
length (Optional[int]) – length hint to provide
length_factory (Optional[Callable[[], int]]) – a callable called with no arguments to get the length
You must provide either length or length_factory. Giving them both is wrong, and will result in ValueError
- close()¶
- generator¶
- length¶
- length_factory¶
- send(obj)¶
- satella.coding.iterators.run_when_iterator_completes(iterator, func_to_run, do_exception=None, *args, **kwargs)¶
Schedule a function to be called when an iterator completes.
- Parameters:
iterator (Iterator) – iterator to use
func_to_run (Callable) – function to run afterwards, but only if there were no exceptions or they were swallowed by do_exception.
do_exception – a callable to call with the exception instance if generator fails at some point. Note that if this doesn’t re-raise the exception, it will be swallowed. Default behaviour is just to re-raise it.
args – arguments to pass to the function
kwargs – keyword arguments to pass to the function
satella.coding.metaclasses module¶
- satella.coding.metaclasses.CopyDocsFrom(target_cls)¶
A metaclass to copy documentation from some other class for respective methods.
>>> class Source: >>> def test(self): >>> 'docstring' >>> class Target(metaclass=CopyDocsFrom(Source)): >>> def test(self): >>> ... >>> assert Target.test.__doc__ == Source.test.__doc__
- Parameters:
target_cls (Type) – class from which to copy the docs
- satella.coding.metaclasses.DocsFromParent(name, bases, dictionary)¶
A metaclass that fetches missing docstring’s for methods from the classes’ bases, looked up BFS. This will fetch the class’s docstring itself, if available and not present in the child.
>>> class Father: >>> def test(self): >>> '''my docstring'''
>>> class Child(Father, metaclass=DocsFromParent): >>> def test(self): >>> ... >>> assert Child.test.__doc__ == 'my docstring'
- Parameters:
name (str) –
bases (Tuple[type]) –
dictionary (dict) –
- Return type:
Type
- satella.coding.metaclasses.dont_wrap(fun)¶
A special decorator to save given class member from being mulched by wrap_with
- satella.coding.metaclasses.get_noconflict_metaclass(bases, left_metas, right_metas)¶
Not intended to be used outside of this module, unless you know what you are doing.
- Return type:
Callable[[str, tuple, dict], Type]
- satella.coding.metaclasses.metaclass_maker(name, bases, a_dict)¶
Automatically construct a compatible meta-class like interface. Use like:
>>> class C(A, B, metaclass=metaclass_maker): >>> pass
- Parameters:
name (str) –
bases (tuple) –
a_dict (dict) –
- Return type:
Type
- satella.coding.metaclasses.metaclass_maker_f(left_metas=(), right_metas=())¶
- satella.coding.metaclasses.remove_redundant(metaclasses)¶
- satella.coding.metaclasses.skip_redundant(iterable, skip_set=None)¶
Redundant items are repeated items or items in the original skip_set.
- satella.coding.metaclasses.wrap_property(getter=<function <lambda>>, setter=<function <lambda>>, deleter=<function <lambda>>)¶
Construct a property wrapper.
This will return a function, that if given a property, will wrap it’s getter, setter and deleter with provided functions.
Getter, setter and deleter are extracted from fget, fset and fdel, so only native properties, please, not descriptor-objects.
- Parameters:
getter (Callable[[Callable[[object], Any]], Callable[[object], Any]]) – callable that accepts a callable(instance) -> value, and returns the same. Getter will be wrapped by this
setter (Callable[[Callable[[object, Any], None]], Callable[[object, Any], None]]) – callable that accepts a callable(instance, value) and returns the same. Setter will be wrapped by this
deleter (Callable[[Callable[[object], None]], Callable[[object], None]]) – callable that accepts a callable(instance), and returns the same. Deleter will be wrapped by this
- satella.coding.metaclasses.wrap_with(callables=<function <lambda>>, properties=<function <lambda>>, selector_callables=<function <lambda>>, selector_properties=<function <lambda>>)¶
A metaclass that wraps all elements discovered in this class with something
Example:
>>> def make_double(fun): >>> return lambda self, x: fun(x)*2 >>> class Doubles(metaclass=wrap_all_methods_with(make_double)): >>> def return_four(self, x): >>> return 2 >>> assert Doubles().return_four(4) == 4
Note that every callable that appears in the class namespace, ie. object that has __call__ will be considered for wrapping.
This is compatible with the abc.ABCMeta metaclass
- Parameters:
callables (Callable[[Callable], Callable]) – function to wrap all callables with given class with
properties (Callable[[property], property]) – function to wrap all properties with given class with
selector_callables (Callable[[Callable], bool]) – additional criterion to be ran on given callable before deciding to wrap it. It must return True for wrapping to proceed.
selector_properties (Callable[[property], bool]) – additional criterion to be ran on given property before deciding to wrap it. It must return True for wrapping to proceed.
satella.coding.misc module¶
- class satella.coding.misc.Closeable¶
Bases:
object
A class that needs to clean up its own resources.
It’s destructor calls .close(). Use like this:
>>> class MyClose(Closeable): >>> def close(self): >>> if super().close(): >>> .. clean up ..
Can be also used as a context manager, with close() called upon __exit__.
Warning
You should extend both __init__ and close(). Invoke __init__() at the end of your class constructor, this will prevent the destructor from closing on half-initialized classes.
Objects before initialization (calling of this constructor) are considered closed. Checking if they are closed will emit a warning.
- close()¶
Check if the resource needs cleanup, and clean up this resource.
Use like this:
>>> class MyClose(Closeable): >>> def close(self): >>> if super().close(): >>> .. clean up ..
- Returns:
whether the cleanup should proceed
- Raises:
RuntimeError – the constructor was not invoked
- Return type:
bool
- property closed: bool¶
- Returns:
whether this object is closed
- satella.coding.misc.call_with_arguments(function, arguments)¶
Call a function, but with giving it arguments via a dictionary.
Dictionary should be a mapping of argument name to it’s value.
- Parameters:
function (Callable) – function to call
arguments (Dict[str, Any]) – a dict of arguments : argument name => argument value. This dictionary will be modified!
- Returns:
return value of the function
- Raises:
TypeError – too few arguments, or some arguments required were missing
ValueError – too many arguments given
- Return type:
Any
- satella.coding.misc.chain_callables(callable1, callable2)¶
Link two callables together. callable2, if it takes an argument, will receive callables’1 result, and if it takes no arguments it will received nothing.
- Parameters:
callable1 (Callable) – first callable to call
callable2 (Callable) – callable to call with callable1’s result
- Returns:
result of callable2
- Return type:
Callable
- satella.coding.misc.contains(needle, haystack)¶
A syntactic sugar for the following:
>>> for item in haystack: >>> if needle == item: >>> return True >>> return False
Note that this is very like Python’s in operator, however it’s not quite same, since in doesn’t involve the __eq__ operator at every step!
This function for example allows you to circumvent Python’s limitations concerning
ComparableEnum
- Parameters:
needle – needle to check for
haystack – haystack to check against
- Returns:
whether haystack contains the element
- Return type:
bool
- satella.coding.misc.enum_value(value)¶
If value is an enum, extract and return it’s value.
Otherwise, return it as-is.
- Parameters:
value – value to extract enum from
- Returns:
value
- satella.coding.misc.get_arguments(function, *args, **kwargs)¶
Return local variables that would be defined for given function if called with provided arguments.
Note that this function will not return the “self” argument of methods and it won’t return the class of “cls” of classmethods.
- Parameters:
function (Callable) – callable to examine
args – arguments to provide
kwargs – keyword arguments to provide
- Returns:
a dictionary of local variables with their values, as they would appear in function if called with provided arguments
- Raises:
TypeError – the dictionary cannot be created with provided arguments
- Return type:
Dict[str, Any]
- satella.coding.misc.length(lenable)¶
Return length of an item. If it is a generator, exhaust it and return it’s length.
- Return type:
int
- satella.coding.misc.queue_iterator(queue)¶
Syntactic sugar for
>>> while queue.qsize() > 0: >>> yield queue.get()
- Parameters:
queue (Queue) –
- Return type:
Iterator
- satella.coding.misc.source_to_function(src)¶
If src is callable, return it as-is Transform a string containing a Python expression with a variable x to a lambda.
It will be treated as if it was appended to ‘lambda x: ‘
WARNING: Do not run untrusted data. Familiarize yourself with the dangers of passing unvalidated data to exec() or eval()!
- Parameters:
src (Union[Callable, str]) – a callable or a Python string expression
- Returns:
a callable
- Return type:
Callable[[Any], Any]
- satella.coding.misc.update_attr_if_none(obj, attr, value, on_attribute_error=True, if_value_is_not_none=False)¶
Updates the object attribute, if it’s value is None, or if it yields AttributeError (customizable as per on_attribute_error parameter)
- Parameters:
obj (object) – object to alter
attr (str) – attribute to set
value (Any) – value to set
on_attribute_error (bool) – whether to proceed with setting the value on AttributeError while trying to read given attribute. If False, AttributeError will be raised.
if_value_is_not_none (bool) – update object unconditionally, if only value is not None
- Returns:
obj
- Return type:
object
- satella.coding.misc.update_if_not_none(dictionary, key, value)¶
Deprecated alias for
update_key_if_none()
Deprecated since version 2.14.22.
- Parameters:
dictionary (Dict) –
key (Hashable) –
- Return type:
Dict
- satella.coding.misc.update_key_if_none(dictionary, key, value)¶
This is deprecated. Please use update_key_if_not_none instead!
Deprecated since version 2.14.22.
- Parameters:
dictionary (Dict) –
key (Hashable) –
- Return type:
Dict
- satella.coding.misc.update_key_if_not_none(dictionary, key, value=<class 'satella.coding.misc._BLANK'>)¶
Syntactic sugar for
>>> if value is not None: >>> dictionary[key] = value
If value is passed, else
>>> for key, value in key.items(): >>> if value is not None: >>> dictionary[key] = value
- Parameters:
dictionary (Dict) – dictionary to update
key (Union[Hashable, Dict]) – key to use or a dictionary of items
value (Union[Any, Type[_BLANK]]) – value to use
- Returns:
the dictionary itself
- Return type:
Dict
- satella.coding.misc.update_key_if_true(dictionary, key, value, flag=<class 'satella.coding.misc._BLANK'>)¶
If flag is True, execute dictionary[key] = value
- Parameters:
dictionary (Dict) – dictionary to mutate
key (Hashable) – dictionary key to use
value (Any) – dictionary value to set
flag (Union[bool, Type[_BLANK]]) – whether to execute the setting operation. If let at default, flag will be calculated from boolean of the value
- Returns:
the dict itself
- Return type:
Dict
satella.coding.optionals module¶
- class satella.coding.optionals.Optional(obj)¶
Bases:
Proxy
[T
]A wrapper for your classes, that does nothing if the object passed in is None. It will return an empty Optional in that case.
Usage example:
>>> may_be_none = None >>> Optional(may_be_none).cancel().result()
So far operations supported:
calling
getattr
getitem/setitem/delitem
testing for truth
comparison (with nonexistent elements always comparing false)
membership test (with nonexistent elements always returning false)
Warning
Returned objects via getattr and getitem are NOT wrapped in an Optional. You need to do it by hand or just file an issue. I’ll add that when I need it.
- Parameters:
obj – object to wrap
- satella.coding.optionals.call_if_nnone(clbl, *args, **kwargs)¶
Call clbl with provided arguments, but only if clbl is not None.
If it’s None, then None will be returned. Else, the result of this callable will be returned.
- Parameters:
clbl (Optional[Callable[[...], V]]) – callable to call, or a None
args – positional arguments to provide to clbl
kwargs – keyword arguments to provide to clbl
- Returns:
return value or None
- Return type:
Optional[V]
- satella.coding.optionals.extract_optional(v)¶
If v is an optional, extract the value that it wraps. If it is not, return v
- Parameters:
v (Union[T, Union[T]]) – value to extract the value from
- Returns:
resulting value
- Return type:
T
- satella.coding.optionals.iterate_if_nnone(iterable)¶
Return a generator iterating over every element of iterable if it’s not None.
If it’s None, return an empty generator
- Parameters:
iterable (Optional[Iterable]) – iterable to iterate over
- Returns:
an empty generator if iterable is none, else an iterator over iterable
- Return type:
Iterable
satella.coding.overloading module¶
- class satella.coding.overloading.TypeSignature(t_sign)¶
Bases:
Signature
A type signature.
You can compare signatures:
>>> def a(y: object): >>> pass >>> def b(y: int): >>> pass >>> TypeSignature.from_fun(a) < TypeSignature(b)
- Parameters:
t_sign (inspect.Signature) – a inspect.Signature
- can_be_called_with_args(*args, **kwargs)¶
Can this type signature be called with following arguments?
- Return type:
bool
- static from_fun(fun)¶
Return a type signature from a function
- Return type:
- is_more_generic_than(b)¶
Is this type signature more generic than an other?
- Parameters:
b (TypeSignature) –
- Return type:
bool
- matches(*args, **kwargs)¶
Does this invocation match this signature?
- Return type:
bool
- class satella.coding.overloading.class_or_instancemethod¶
Bases:
classmethod
A decorator to make your methods both classmethods (they will receive an instance of type as their first argument) or normal methods (they will receive an instance of their type).
Use like:
>>> class MyClass: >>> @class_or_instancemethod >>> def method(self_or_cls): >>> if isinstance(self_or_cls, MyClass): >>> # method code >>> else: >>> # classmethod code
- class satella.coding.overloading.overload(fun)¶
Bases:
object
A class used for method overloading.
Warning
This feature is scheduled for an overhaul and may not work as promised. Keep that in mind.
Note that methods can be only overloaded by their positional, or positional-and-keyword arguments. Overload distinguishment will be done at the level of positional arguments only.
Note that typing checks will be done via isinstance().
Use like this:
>>> @overload >>> def what_type(x: str): >>> print('String') >>> @what_type.overload >>> def what_type(x: int): >>> print('Int') >>> what_type(5) >>> what_type('string')
Note that this instance’s __wrapped__ will refer to the first function. TypeError will be called if no signatures match arguments.
- Parameters:
fun (tp.Callable) –
- property all_functions: Iterable[object]¶
Return a list of all functions registered within this overload
- overload(fun)¶
- Raises:
ValueError – this signature already has an overload
satella.coding.predicates module¶
- class satella.coding.predicates.PredicateClass(operation=<function _nop>)¶
Bases:
object
A shorthand to create lambdas using such statements, for example:
>>> add_two = x + 2 >>> assert add_two(2) == 4
- Parameters:
operation (Callable[[Any], Any]) –
- false()¶
Return a predicate checking whether value is False
- Return type:
Callable[[T], bool]
- float()¶
Call float() on predicate
- Return type:
Callable[[T], bool]
- has(predicate)¶
Check if any element of the current value (which must be an iterable) returns True when applied to predicate
- Parameters:
predicate (PredicateClass) – predicate that has to return True for at least one of this predicate’s values
- Return type:
Callable[[T], bool]
- has_keys(*keys)¶
Return a predicate checking whether this value has provided keys
- Return type:
Callable[[T], bool]
- has_p(predicate)¶
An old name for has().
It’s deprecated. Use has() instead
Deprecated since version 2.14.22.
- Parameters:
predicate (PredicateClass) –
- Return type:
Callable[[T], bool]
- identity()¶
Spawn another object with the same operation, but different identity.
Used for constructing dicts keyed by predicates.
- Return type:
Callable[[T], bool]
- inside(a)¶
Return a predicate checking if x is inside value
- Parameters:
a (Callable) –
- Return type:
Callable[[T], bool]
- instanceof(a)¶
Return a predicate checking whether this value is an instance of instance
- Parameters:
a (Callable) –
- Return type:
Callable[[T], bool]
- int()¶
Call int() on predicate
- Return type:
Callable[[T], bool]
- is_instance(*args)¶
Check if given value is one of instances.
- Parameters:
args – will be passed as argument to isinstance
- is_valid_schema(schema=None, **kwargs)¶
Check if given value has the correct schema. The schema is the same as in
is_valid_schema()
- Parameters:
schema (Optional[Union[Descriptor, Dict]]) –
- length()¶
Return a predicate returning length of its argument
- Return type:
Callable[[T], bool]
- one_of(*values)¶
Return a predicate checking if x is amongst values
- Return type:
Callable[[T], bool]
- operation¶
- str()¶
Call str() on predicate
- Return type:
Callable[[T], bool]
- type()¶
Return a predicate returning the type of it’s argument
- Return type:
Type
- satella.coding.predicates.build_structure(struct, argument, final_operator=<function <lambda>>, nested_call=False)¶
Given a structure (tuple, list, dict) that contains x’s as some of the elements, build such a structure corresponding to given that all x’s are replaced by result of their calculation on argument.
Just note that if you’re constructing dictionaries, use the .identity() method of predicate, to randomize it’s identity.
- Parameters:
struct (Union[tuple, list, dict]) – structure to build
argument – argument
final_operator – an operator to call on the result
nested_call – internal, don’t use
- Returns:
analogous structure
- Return type:
Union[tuple, list, dict]
satella.coding.recast_exceptions module¶
- satella.coding.recast_exceptions.catch_exception(exc_class, clb, return_instead=None, return_value_on_no_exception=False)¶
Catch exception of given type and return it. Functionally equivalent to:
>>> try: >>> v = clb() >>> if return_value_on_no_exception: >>> return v >>> except exc_class as e: >>> if return_instead: >>> return return_instead >>> return e
If a different class of exception is caught, it will be propagated.
- Parameters:
exc_class (Union[Type[Exception], Tuple[Type[Exception], ...]]) – Exception classes to catch
clb (Callable[[], Optional[T]]) – callable/0 to call to raise the exception
return_instead (Optional[T]) – what to return instead of the function result if it didn’t end in an exception
return_value_on_no_exception (bool) – whether to return the function result if exception didn’t happen
- Raises:
ValueError – an exception was not thrown
- Return type:
Union[Exception, T]
- class satella.coding.recast_exceptions.log_exceptions(logger, severity=40, format_string='{e}', locals_=None, exc_types=<class 'Exception'>, swallow_exception=False)¶
Bases:
object
Decorator/context manager to log your exceptions into the log.
The exception will be logged and re-raised.
Logger will be passed the exception instance as exc_info.
- Parameters:
logger (Logger) – a logger to which the exception has to be logged
severity (int) – a severity level
format_string (str) –
a format string with fields: - e : the exception instance itself - args : positional arguments with which the function was called, unavailable if context
manager
- kwargskeyword arguments with which the function was called, unavailable if context
manager
You can specify additional fields providing the locals_ argument Example: “{exc_type} occurred with message {exc_val} with traceback {exc_tb}”
locals – local variables to add to the format string. args and kwargs will be overwritten by this, but e will never be overwritten.
exc_types (Union[Type[Exception], Sequence[Type[Exception]]]) – logger will log only on those exceptions. Default is None which means log on all exceptions
swallow_exception (bool) – if True, exception will be swallowed
locals_ (Optional[Dict]) –
- analyze_exception(e, args, kwargs)¶
Return whether the exception has been logged
- Return type:
bool
- exc_types¶
- format_string¶
- locals¶
- logger¶
- severity¶
- swallow_exception¶
- satella.coding.recast_exceptions.raises_exception(exc_class, clb)¶
Does the callable raise a given exception?
- Parameters:
exc_class (Union[Type[Exception], Tuple[Type[Exception], ...]]) –
clb (Callable[[], None]) –
- Return type:
bool
- class satella.coding.recast_exceptions.reraise_as(source_exc, target_exc, *args, **kwargs)¶
Bases:
object
Transform some exceptions into others.
Either a decorator or a context manager
New exception will be created by calling exception to transform to with repr of current one.
You can also provide just two exceptions, eg.
>>> reraise_as(NameError, ValueError, 'a value error!')
You can also provide a catch-all:
>>> reraise_as((NameError, ValueError), OSError, 'an OS error!')
New exception will be raised from the one caught!
Note
This checks if exception matches directly via
isinstance
, so defining your own subclassing hierarchy by__isinstance__
or__issubclass__
will work here.This is meant as an improvement of
rethrow_as
- Parameters:
source_exc (Union[Type[Exception], Tuple[Type[Exception], ...]]) – source exception or a tuple of exceptions to catch
target_exc (Optional[Type[Exception]]) – target exception to throw. If given a None, the exception will be silently swallowed.
args – arguments to constructor of target exception
kwargs – keyword arguments to constructor of target exception
- args¶
- kwargs¶
- source¶
- target_exc¶
- class satella.coding.recast_exceptions.rethrow_as(*pairs, exception_preprocessor=<built-in function repr>, returns=None, returns_factory=None)¶
Bases:
object
Transform some exceptions into others.
Either a decorator or a context manager
New exception will be created by calling exception to transform to with repr of current one.
Note
This checks if exception matches directly via
isinstance
, so defining your own subclassing hierarchy by__isinstance__
or__issubclass__
will work here.You can also provide just two exceptions, eg.
>>> rethrow_as(NameError, ValueError)
You can also provide a pairwise translation, eg. from NameError to ValueError and from OSError to IOError
>>> rethrow_as((NameError, ValueError), (OSError, IOError))
If the second value is a None, exception will be silenced.
Pass tuples of (exception to catch - exception to transform to).
Warning
Try to use
reraise_as
instead. However, during to richer set of switches and capability to return a value this is not deprecated.Deprecated since version v.
- Parameters:
exception_preprocessor (Optional[Callable[[Exception], str]]) – other callable/1 to use instead of repr. Should return a str, a text description of the exception
returns – what value should the function return if this is used as a decorator
returns_factory (Optional[Callable[[], Any]]) – a callable that returns the value this function should return is this is used as as decorator
pairs (Union[Type[Exception], Tuple[Type[Exception], ...]]) –
- Raises:
ValueError – you specify both returns and returns_factory
- exception_preprocessor¶
- mapping¶
- returns¶
- returns_factory¶
- satella.coding.recast_exceptions.silence_excs(*exc_types, returns=None, returns_factory=None)¶
Silence given exception types.
Can be either a decorator or a context manager.
If you are using it as a decorator, you can specify what value should the function return by using the returns kwarg:
>>> @silence_excs(KeyError, returns=5) >>> def returns_5(): >>> raise KeyError() >>> assert returns_5() == 5
Or if you want to you can specify a callable that will return the value you want to return
>>> @silence_excs(KeyError, returns_factory=lambda: 5) >>> def returns_5(): >>> raise KeyError() >>> assert returns_5() == 5
- Raises:
ValueError – you gave both returns and returns_factory. You can only pass one of them!
- Parameters:
exc_types (Type[Exception]) –
returns_factory (Optional[Callable[[], Any]]) –
satella.coding.typing module¶
- class satella.coding.typing.Appendable(*args, **kwargs)¶
Bases:
Protocol
[T
]- append(item)¶
- Parameters:
item (T) –
- Return type:
None
- class satella.coding.typing.NoneType¶
Bases:
object
Module contents¶
Just useful objects to make your coding nicer every day
- class satella.coding.Closeable¶
Bases:
object
A class that needs to clean up its own resources.
It’s destructor calls .close(). Use like this:
>>> class MyClose(Closeable): >>> def close(self): >>> if super().close(): >>> .. clean up ..
Can be also used as a context manager, with close() called upon __exit__.
Warning
You should extend both __init__ and close(). Invoke __init__() at the end of your class constructor, this will prevent the destructor from closing on half-initialized classes.
Objects before initialization (calling of this constructor) are considered closed. Checking if they are closed will emit a warning.
- close()¶
Check if the resource needs cleanup, and clean up this resource.
Use like this:
>>> class MyClose(Closeable): >>> def close(self): >>> if super().close(): >>> .. clean up ..
- Returns:
whether the cleanup should proceed
- Raises:
RuntimeError – the constructor was not invoked
- Return type:
bool
- property closed: bool¶
- Returns:
whether this object is closed
- class satella.coding.Context(parent=None, **variables)¶
Bases:
object
New layer of environment. Can have it’s own variables, or can hoist them onto the parent.
Warning
This is considered experimental. I just haven’t found out a good use case for it yet.
- Parameters:
parent (tp.Optional[Context]) –
- does_exist(val)¶
Does a given value exist on stack for this call of function?
- Parameters:
val (str) –
- Return type:
bool
- push_up(item, value=<object object>)¶
Advance current variable to the top of the card stack.
- Parameters:
item (str) – variable name
value – if not given, current value of given variable will be taken
- Return type:
None
- satella.coding.CopyDocsFrom(target_cls)¶
A metaclass to copy documentation from some other class for respective methods.
>>> class Source: >>> def test(self): >>> 'docstring' >>> class Target(metaclass=CopyDocsFrom(Source)): >>> def test(self): >>> ... >>> assert Target.test.__doc__ == Source.test.__doc__
- Parameters:
target_cls (Type) – class from which to copy the docs
- class satella.coding.DictDeleter(dict_to_process)¶
Bases:
object
Having problems deleting entries from your dict while iterating on them? No problem. Just swap the following:
>>> keys_to_delete = [] >>> for key, value in my_dict.items(): >>> if value.should_delete(): >>> keys_to_delete.append(key) >>> for key in keys_to_delete: >>> del my_dict[key]
With the following:
>>> with DictDeleter(my_list) as ld: >>> for key, value in ld.items(): >>> if value.should_delete(): >>> ld.delete()
Note that a single DictDeleter running from a single context must be iterated on by only a single Thread as it keeps the state of iterator in itself, to prevent allocating new objects and slowing things down too much.
This allocates only a single object per a call to delete().
- Parameters:
dict_to_process (MutableMapping) –
- current_iterator¶
- current_key¶
- delete()¶
- Return type:
None
- dict_to_process¶
- items()¶
- Return type:
- iter_mode¶
- keys()¶
- Return type:
- keys_to_delete¶
- values()¶
- Return type:
- satella.coding.DocsFromParent(name, bases, dictionary)¶
A metaclass that fetches missing docstring’s for methods from the classes’ bases, looked up BFS. This will fetch the class’s docstring itself, if available and not present in the child.
>>> class Father: >>> def test(self): >>> '''my docstring'''
>>> class Child(Father, metaclass=DocsFromParent): >>> def test(self): >>> ... >>> assert Child.test.__doc__ == 'my docstring'
- Parameters:
name (str) –
bases (Tuple[type]) –
dictionary (dict) –
- Return type:
Type
- class satella.coding.EmptyContextManager(*args, **kwargs)¶
Bases:
object
A context manager that does nothing. Only to support conditional change of context managers, eg in such a way:
>>> if tx_enabled: >>> ctxt = transaction.atomic >>> else: >>> ctxt = EmptyContextManager() >>> with ctxt: >>> ...
Note that it will accept any parameters, and then throw them on the ground.
- exception satella.coding.Inequal(obj1, obj2, reason)¶
Bases:
Exception
An exception raised by
deep_compare()
if two objects don’t match- Variables:
obj1 – first object that was not equal, or key name
obj2 – second object that was not equal, or None
reason – (
InequalityReason
) reason for inequality
- Parameters:
reason (InequalityReason) –
- class satella.coding.InequalityReason(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)¶
Bases:
IntEnum
- KEY_NOT_FOUND = 2¶
key given as obj1 was not found
- LENGTH_MISMATCH = 1¶
length didn’t match
- NOT_EQUAL = 0¶
direct eq yielded not equal
- class satella.coding.ListDeleter(list_to_process, direction=0)¶
Bases:
Generic
[T
]Having problems deleting entries from your list while iterating on them? No problem. Just swap the following:
>>> entries_to_delete = [] >>> for entry in my_list: >>> if entry.should_delete(): >>> entries_to_delete.append(entry) >>> for entry in entries_to_delete: >>> my_list.remove(entry)
With the following:
>>> with ListDeleter(my_list) as ld: >>> for entry in ld: >>> if entry.should_delete(): >>> ld.delete()
You can also use the alternative syntax of: >>> ld = ListDeleter(my_list) >>> while True: >>> try: >>> v = ld.next() >>> except StopIteration: >>> break >>> if condition(v): >>> ld.delete() >>> ld.remove_items()
Note that a single ListDeleter running from a single context must be iterated on by only a single Thread as it keeps the state of iterator in itself, to prevent allocating new objects and slowing things down too much.
Note that calling reversed() on this will reset the pointer to the end of the list or the beginning of the list, respectively.
This allocates only a single object per a call to delete().
Calling the list deleter during iteration will yield the element.
You can pass any type of object here, as long as it supports pop(position) and __getitem__
- Parameters:
list_to_process (MutableSequence[T]) –
direction (int) –
- current_index¶
- delete()¶
- Return type:
None
- direction¶
- indices_to_delete: Set[int]¶
- list_to_process¶
- next()¶
- Returns:
the next element
- Raises:
StopIteration – no more entries
- Return type:
T
- prev()¶
Move to previous element, as per ordering.
- Returns:
the previous element
- Raises:
StopIteration – list is already at the first element!
- Return type:
T
- remove_items()¶
After all of the items have been marked for deletion, delete them
- Return type:
None
- removed¶
- property value: T¶
- class satella.coding.Monitor¶
Bases:
object
Base utility class for creating monitors (the synchronization thingies!)
These are NOT re-entrant!
Use it like that:
>>> class MyProtectedObject(Monitor): >>> def __init__(self, *args, **kwargs): >>> Monitor.__init__(self) >>> ... do your job ..
>>> @Monitor.synchronized >>> def function_that_needs_mutual_exclusion(self): >>> .. do your threadsafe jobs ..
>>> def function_that_partially_needs_protection(self): >>> .. do your jobs .. >>> with Monitor.acquire(self): >>> .. do your threadsafe jobs .. >>> .. do your jobs .. >>> with self: >>> .. do your threadsafe jobs ..
You need to invoke this at your constructor You can also use it to release locks of other objects.
- class acquire(foo)¶
Bases:
object
Returns a context manager object that can lock another object, as long as that object is a monitor.
Consider foo, which is a monitor. If you needed to lock it from outside, you would do:
>>> with Monitor.acquire(foo): >>> .. do operations on foo that need mutual exclusion ..
- Parameters:
foo (Monitor) –
- foo¶
- class release(foo)¶
Bases:
object
Returns a context manager object that can release another object as long as that object is a monitor.
Consider foo, which is a monitor. You have a protected function, but you feel that you can release it for a while as it would improve parallelism. You can use it as such:
>>> @Monitor.synchronized >>> def protected_function(self): >>> .. do some stuff that needs mutual exclusion .. >>> with Monitor.release(self): >>> .. do some I/O that does not need mutual exclusion .. >>> .. back to protected stuff ..
- Parameters:
foo (Monitor) –
- foo¶
- classmethod synchronize_on(monitor)¶
A decorator for locking on non-self Monitor objects
Use it like:
>>> class MasterClass(Monitor): >>> def get_object(self): >>> class SlaveClass: >>> @Monitor.synchronize_on(self) >>> def get_object(self2): >>> ... >>> return SlaveClass
- Parameters:
monitor (Monitor) –
- Return type:
Callable[[Callable], Callable]
- static synchronize_on_attribute(attr_name)¶
When a Monitor is an attribute of a class, and you have a method instance that you would like secure by acquiring that monitor, use this.
The first argument taken by that method instance must be self.
- Parameters:
attr_name (str) – name of the attribute that is the monitor
- static synchronized(fun)¶
This is a decorator. Class method decorated with that will lock the global lock of given instance, making it threadsafe. Depending on usage pattern of your class and it’s data semantics, your performance may vary
- Parameters:
fun (Callable) –
- Return type:
Callable
- class satella.coding.RMonitor¶
Bases:
Monitor
Monitor, but using an reentrant lock instead of a normal one
You need to invoke this at your constructor You can also use it to release locks of other objects.
- class satella.coding.RunActionAfterGeneratorCompletes(generator, *args, call_despite_closed=False, **kwargs)¶
Bases:
Generator
Run an action after a generator completes. An abstract class.
Please note that this routine will be called only when the generator completes. If you abort it prematurely, via close()
- Parameters:
generator (Generator) – generator to watch for
args – arguments to invoke action_to_run with
call_despite_closed (bool) –
action_to_run()
will be called even if the generator is closedkwargs – keyword arguments to invoke action_to_run with
- abstract action_to_run(*args, **kwargs)¶
This will run when this generator completes. Override it.
- args¶
- call_despite_closed¶
- call_on_exception(exc)¶
This will run when this generator throws any exception inside it’s __next__() or send(). You can reraise it (which is the default behavior if you do not override this).
- Parameters:
exc (Exception) –
- close()¶
Close this generator. Note that this will cause
action_to_run()
not to run
- closed¶
- generator¶
- kwargs¶
- next()¶
- send(value)¶
Send a value to the generator
- throw(_RunActionAfterGeneratorCompletes__typ, _RunActionAfterGeneratorCompletes__val=None, _RunActionAfterGeneratorCompletes__tb=None)¶
Raise an exception in the generator. Return next yielded value or raise StopIteration.
- class satella.coding.SelfClosingGenerator(generator)¶
Bases:
object
A wrapper to exhaust the generator in response to closing it.
This will allow generators to complete that don’t provide a .close() method.
This will additionally exhaust the generator upon deallocation of the generator.
You can feed it with either generators, or generator-functions, it will behave correctly each time.
You can also use it as a context manager, to decouple finalizing the generator from the GC collection
- Parameters:
generator (Union[Generator, Callable[[Any], Generator]]) –
- close()¶
- Return type:
None
- generator¶
- send(obj)¶
- Parameters:
obj (Any) –
- Return type:
None
- stopped: bool¶
- class satella.coding.TypeSignature(t_sign)¶
Bases:
Signature
A type signature.
You can compare signatures:
>>> def a(y: object): >>> pass >>> def b(y: int): >>> pass >>> TypeSignature.from_fun(a) < TypeSignature(b)
- Parameters:
t_sign (inspect.Signature) – a inspect.Signature
- can_be_called_with_args(*args, **kwargs)¶
Can this type signature be called with following arguments?
- Return type:
bool
- static from_fun(fun)¶
Return a type signature from a function
- Return type:
- is_more_generic_than(b)¶
Is this type signature more generic than an other?
- Parameters:
b (TypeSignature) –
- Return type:
bool
- matches(*args, **kwargs)¶
Does this invocation match this signature?
- Return type:
bool
- satella.coding.assert_equal(a, b)¶
Assert that two values are equal. If not, an
satella.coding.Inequality
exception will be thrown.Objects are tried to compare using it’s
__eq__
.- Parameters:
a – first value to compare
b – second value to compare
- Raises:
Inequal – objects were not equal
- 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.
- 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.call_with_arguments(function, arguments)¶
Call a function, but with giving it arguments via a dictionary.
Dictionary should be a mapping of argument name to it’s value.
- Parameters:
function (Callable) – function to call
arguments (Dict[str, Any]) – a dict of arguments : argument name => argument value. This dictionary will be modified!
- Returns:
return value of the function
- Raises:
TypeError – too few arguments, or some arguments required were missing
ValueError – too many arguments given
- Return type:
Any
- satella.coding.catch_exception(exc_class, clb, return_instead=None, return_value_on_no_exception=False)¶
Catch exception of given type and return it. Functionally equivalent to:
>>> try: >>> v = clb() >>> if return_value_on_no_exception: >>> return v >>> except exc_class as e: >>> if return_instead: >>> return return_instead >>> return e
If a different class of exception is caught, it will be propagated.
- Parameters:
exc_class (Union[Type[Exception], Tuple[Type[Exception], ...]]) – Exception classes to catch
clb (Callable[[], Optional[T]]) – callable/0 to call to raise the exception
return_instead (Optional[T]) – what to return instead of the function result if it didn’t end in an exception
return_value_on_no_exception (bool) – whether to return the function result if exception didn’t happen
- Raises:
ValueError – an exception was not thrown
- Return type:
Union[Exception, T]
- satella.coding.chain(*args)¶
Construct an iterator out of provided elements.
If an element is an iterator, or an iterable it will be yielded-from. If it’s not, it will just be yielded.
A cast to iter() is used to determine iteratorness
- Return type:
Iterator
- satella.coding.chain_callables(callable1, callable2)¶
Link two callables together. callable2, if it takes an argument, will receive callables’1 result, and if it takes no arguments it will received nothing.
- Parameters:
callable1 (Callable) – first callable to call
callable2 (Callable) – callable to call with callable1’s result
- Returns:
result of callable2
- Return type:
Callable
- satella.coding.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
- class satella.coding.class_or_instancemethod¶
Bases:
classmethod
A decorator to make your methods both classmethods (they will receive an instance of type as their first argument) or normal methods (they will receive an instance of their type).
Use like:
>>> class MyClass: >>> @class_or_instancemethod >>> def method(self_or_cls): >>> if isinstance(self_or_cls, MyClass): >>> # method code >>> else: >>> # classmethod code
- satella.coding.contains(needle, haystack)¶
A syntactic sugar for the following:
>>> for item in haystack: >>> if needle == item: >>> return True >>> return False
Note that this is very like Python’s in operator, however it’s not quite same, since in doesn’t involve the __eq__ operator at every step!
This function for example allows you to circumvent Python’s limitations concerning
ComparableEnum
- Parameters:
needle – needle to check for
haystack – haystack to check against
- Returns:
whether haystack contains the element
- Return type:
bool
- satella.coding.dont_wrap(fun)¶
A special decorator to save given class member from being mulched by wrap_with
- satella.coding.enum_value(value)¶
If value is an enum, extract and return it’s value.
Otherwise, return it as-is.
- Parameters:
value – value to extract enum from
- Returns:
value
- satella.coding.exhaust(iterator)¶
Iterate till the end of the iterator, discarding values as they go
- Parameters:
iterator (Iterator) – iterator to exhaust
- Return type:
None
- class satella.coding.expect_exception(exc_to_except, else_raise, *args, **kwargs)¶
Bases:
object
A context manager to use as following:
>>> a = {'test': 2} >>> with expect_exception(KeyError, ValueError, 'KeyError not raised'): >>> a['test2']
If other exception than the expected is raised, it is passed through
- Parameters:
exc_to_except (Union[Type[Exception], Tuple[Type[Exception], ...]]) – a list of exceptions or a single exception to expect
else_raise (Type[Exception]) – raise a particular exception if no exception is raised. This should be a callable that accepts provided args and kwargs and returns an exception instance.
args – args to provide to constructor
kwargs – kwargs to provide to constructor
- else_raise¶
- else_raise_args¶
- else_raise_kwargs¶
- exc_to_except¶
- satella.coding.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.get_arguments(function, *args, **kwargs)¶
Return local variables that would be defined for given function if called with provided arguments.
Note that this function will not return the “self” argument of methods and it won’t return the class of “cls” of classmethods.
- Parameters:
function (Callable) – callable to examine
args – arguments to provide
kwargs – keyword arguments to provide
- Returns:
a dictionary of local variables with their values, as they would appear in function if called with provided arguments
- Raises:
TypeError – the dictionary cannot be created with provided arguments
- Return type:
Dict[str, Any]
- satella.coding.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
- class satella.coding.hint_with_length(generator, length, length_factory=None)¶
Bases:
object
Accepting a generator, return it additionally providing a specified __length_hint__
You can provide generator-generating functions as well
- Parameters:
generator (Generator) – generator to decorate
length (Optional[int]) – length hint to provide
length_factory (Optional[Callable[[], int]]) – a callable called with no arguments to get the length
You must provide either length or length_factory. Giving them both is wrong, and will result in ValueError
- close()¶
- generator¶
- length¶
- length_factory¶
- send(obj)¶
- satella.coding.length(lenable)¶
Return length of an item. If it is a generator, exhaust it and return it’s length.
- Return type:
int
- class satella.coding.log_exceptions(logger, severity=40, format_string='{e}', locals_=None, exc_types=<class 'Exception'>, swallow_exception=False)¶
Bases:
object
Decorator/context manager to log your exceptions into the log.
The exception will be logged and re-raised.
Logger will be passed the exception instance as exc_info.
- Parameters:
logger (Logger) – a logger to which the exception has to be logged
severity (int) – a severity level
format_string (str) –
a format string with fields: - e : the exception instance itself - args : positional arguments with which the function was called, unavailable if context
manager
- kwargskeyword arguments with which the function was called, unavailable if context
manager
You can specify additional fields providing the locals_ argument Example: “{exc_type} occurred with message {exc_val} with traceback {exc_tb}”
locals – local variables to add to the format string. args and kwargs will be overwritten by this, but e will never be overwritten.
exc_types (Union[Type[Exception], Sequence[Type[Exception]]]) – logger will log only on those exceptions. Default is None which means log on all exceptions
swallow_exception (bool) – if True, exception will be swallowed
locals_ (Optional[Dict]) –
- analyze_exception(e, args, kwargs)¶
Return whether the exception has been logged
- Return type:
bool
- exc_types¶
- format_string¶
- locals¶
- logger¶
- severity¶
- swallow_exception¶
- satella.coding.merge_dicts(v1, v2)¶
Try to merge two dicts/list together. If key collision is found, value from v2 will be taken.
If the objects aren’t dicts or lists, v2 will be returned.
Lists will be concatenated, and dicts updated. v1 will be updated in-place!
- Parameters:
v1 (Any) –
v2 (Any) –
- Return type:
Any
- satella.coding.metaclass_maker(name, bases, a_dict)¶
Automatically construct a compatible meta-class like interface. Use like:
>>> class C(A, B, metaclass=metaclass_maker): >>> pass
- Parameters:
name (str) –
bases (tuple) –
a_dict (dict) –
- Return type:
Type
- class satella.coding.overload(fun)¶
Bases:
object
A class used for method overloading.
Warning
This feature is scheduled for an overhaul and may not work as promised. Keep that in mind.
Note that methods can be only overloaded by their positional, or positional-and-keyword arguments. Overload distinguishment will be done at the level of positional arguments only.
Note that typing checks will be done via isinstance().
Use like this:
>>> @overload >>> def what_type(x: str): >>> print('String') >>> @what_type.overload >>> def what_type(x: int): >>> print('Int') >>> what_type(5) >>> what_type('string')
Note that this instance’s __wrapped__ will refer to the first function. TypeError will be called if no signatures match arguments.
- Parameters:
fun (tp.Callable) –
- property all_functions: Iterable[object]¶
Return a list of all functions registered within this overload
- overload(fun)¶
- Raises:
ValueError – this signature already has an overload
- satella.coding.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.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.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.queue_iterator(queue)¶
Syntactic sugar for
>>> while queue.qsize() > 0: >>> yield queue.get()
- Parameters:
queue (Queue) –
- Return type:
Iterator
- satella.coding.raises_exception(exc_class, clb)¶
Does the callable raise a given exception?
- Parameters:
exc_class (Union[Type[Exception], Tuple[Type[Exception], ...]]) –
clb (Callable[[], None]) –
- Return type:
bool
- class satella.coding.rethrow_as(*pairs, exception_preprocessor=<built-in function repr>, returns=None, returns_factory=None)¶
Bases:
object
Transform some exceptions into others.
Either a decorator or a context manager
New exception will be created by calling exception to transform to with repr of current one.
Note
This checks if exception matches directly via
isinstance
, so defining your own subclassing hierarchy by__isinstance__
or__issubclass__
will work here.You can also provide just two exceptions, eg.
>>> rethrow_as(NameError, ValueError)
You can also provide a pairwise translation, eg. from NameError to ValueError and from OSError to IOError
>>> rethrow_as((NameError, ValueError), (OSError, IOError))
If the second value is a None, exception will be silenced.
Pass tuples of (exception to catch - exception to transform to).
Warning
Try to use
reraise_as
instead. However, during to richer set of switches and capability to return a value this is not deprecated.Deprecated since version v.
- Parameters:
exception_preprocessor (Optional[Callable[[Exception], str]]) – other callable/1 to use instead of repr. Should return a str, a text description of the exception
returns – what value should the function return if this is used as a decorator
returns_factory (Optional[Callable[[], Any]]) – a callable that returns the value this function should return is this is used as as decorator
pairs (Union[Type[Exception], Tuple[Type[Exception], ...]]) –
- Raises:
ValueError – you specify both returns and returns_factory
- exception_preprocessor¶
- mapping¶
- returns¶
- returns_factory¶
- satella.coding.run_when_generator_completes(gen, call_on_done, *args, **kwargs)¶
Return the generator with call_on_done to be called on when it finishes
- Parameters:
gen (Generator) – generator
call_on_done (Callable) – callable/0 to call on generator’s completion
args – args to pass to the callable
kwargs – kwargs to pass to the callable
- Returns:
generator
- Return type:
- satella.coding.run_when_iterator_completes(iterator, func_to_run, do_exception=None, *args, **kwargs)¶
Schedule a function to be called when an iterator completes.
- Parameters:
iterator (Iterator) – iterator to use
func_to_run (Callable) – function to run afterwards, but only if there were no exceptions or they were swallowed by do_exception.
do_exception – a callable to call with the exception instance if generator fails at some point. Note that if this doesn’t re-raise the exception, it will be swallowed. Default behaviour is just to re-raise it.
args – arguments to pass to the function
kwargs – keyword arguments to pass to the function
- satella.coding.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.silence_excs(*exc_types, returns=None, returns_factory=None)¶
Silence given exception types.
Can be either a decorator or a context manager.
If you are using it as a decorator, you can specify what value should the function return by using the returns kwarg:
>>> @silence_excs(KeyError, returns=5) >>> def returns_5(): >>> raise KeyError() >>> assert returns_5() == 5
Or if you want to you can specify a callable that will return the value you want to return
>>> @silence_excs(KeyError, returns_factory=lambda: 5) >>> def returns_5(): >>> raise KeyError() >>> assert returns_5() == 5
- Raises:
ValueError – you gave both returns and returns_factory. You can only pass one of them!
- Parameters:
exc_types (Type[Exception]) –
returns_factory (Optional[Callable[[], Any]]) –
- satella.coding.source_to_function(src)¶
If src is callable, return it as-is Transform a string containing a Python expression with a variable x to a lambda.
It will be treated as if it was appended to ‘lambda x: ‘
WARNING: Do not run untrusted data. Familiarize yourself with the dangers of passing unvalidated data to exec() or eval()!
- Parameters:
src (Union[Callable, str]) – a callable or a Python string expression
- Returns:
a callable
- Return type:
Callable[[Any], Any]
- satella.coding.static_var(var_name, starting_value=None)¶
Declare a static variable for given function
Use it like:
>>> @static_var('counter', 2) >>> def count(): >>> count.counter += 1
or:
>>> class MyClass: >>> @static_var('counter', 2) >>> def count(self): >>> MyClass.count.counter += 1
- Parameters:
var_name (str) –
starting_value (Optional[Any]) –
- satella.coding.update_attr_if_none(obj, attr, value, on_attribute_error=True, if_value_is_not_none=False)¶
Updates the object attribute, if it’s value is None, or if it yields AttributeError (customizable as per on_attribute_error parameter)
- Parameters:
obj (object) – object to alter
attr (str) – attribute to set
value (Any) – value to set
on_attribute_error (bool) – whether to proceed with setting the value on AttributeError while trying to read given attribute. If False, AttributeError will be raised.
if_value_is_not_none (bool) – update object unconditionally, if only value is not None
- Returns:
obj
- Return type:
object
- satella.coding.update_if_not_none(dictionary, key, value)¶
Deprecated alias for
update_key_if_none()
Deprecated since version 2.14.22.
- Parameters:
dictionary (Dict) –
key (Hashable) –
- Return type:
Dict
- satella.coding.update_key_if_none(dictionary, key, value)¶
This is deprecated. Please use update_key_if_not_none instead!
Deprecated since version 2.14.22.
- Parameters:
dictionary (Dict) –
key (Hashable) –
- Return type:
Dict
- satella.coding.update_key_if_not_none(dictionary, key, value=<class 'satella.coding.misc._BLANK'>)¶
Syntactic sugar for
>>> if value is not None: >>> dictionary[key] = value
If value is passed, else
>>> for key, value in key.items(): >>> if value is not None: >>> dictionary[key] = value
- Parameters:
dictionary (Dict) – dictionary to update
key (Union[Hashable, Dict]) – key to use or a dictionary of items
value (Union[Any, Type[_BLANK]]) – value to use
- Returns:
the dictionary itself
- Return type:
Dict
- satella.coding.update_key_if_true(dictionary, key, value, flag=<class 'satella.coding.misc._BLANK'>)¶
If flag is True, execute dictionary[key] = value
- Parameters:
dictionary (Dict) – dictionary to mutate
key (Hashable) – dictionary key to use
value (Any) – dictionary value to set
flag (Union[bool, Type[_BLANK]]) – whether to execute the setting operation. If let at default, flag will be calculated from boolean of the value
- Returns:
the dict itself
- Return type:
Dict
- satella.coding.wrap_callable_in_context_manager(clbl, ctxt_mgr, *my_args, **my_kwargs)¶
Wrap a callable in context manager.
Roughly equivalent to:
>>> def inner(*args, **kwargs): >>> with ctxt_mgr(*my_args, **my_kwargs): >>> return clbl(*args, **kwargs) >>> return inner
To be used as:
>>> clbl = wrap_callable_in_context_manager(lambda y: 5, tracing.start_new_span, 'New span')
- satella.coding.wrap_property(getter=<function <lambda>>, setter=<function <lambda>>, deleter=<function <lambda>>)¶
Construct a property wrapper.
This will return a function, that if given a property, will wrap it’s getter, setter and deleter with provided functions.
Getter, setter and deleter are extracted from fget, fset and fdel, so only native properties, please, not descriptor-objects.
- Parameters:
getter (Callable[[Callable[[object], Any]], Callable[[object], Any]]) – callable that accepts a callable(instance) -> value, and returns the same. Getter will be wrapped by this
setter (Callable[[Callable[[object, Any], None]], Callable[[object, Any], None]]) – callable that accepts a callable(instance, value) and returns the same. Setter will be wrapped by this
deleter (Callable[[Callable[[object], None]], Callable[[object], None]]) – callable that accepts a callable(instance), and returns the same. Deleter will be wrapped by this
- satella.coding.wrap_with(callables=<function <lambda>>, properties=<function <lambda>>, selector_callables=<function <lambda>>, selector_properties=<function <lambda>>)¶
A metaclass that wraps all elements discovered in this class with something
Example:
>>> def make_double(fun): >>> return lambda self, x: fun(x)*2 >>> class Doubles(metaclass=wrap_all_methods_with(make_double)): >>> def return_four(self, x): >>> return 2 >>> assert Doubles().return_four(4) == 4
Note that every callable that appears in the class namespace, ie. object that has __call__ will be considered for wrapping.
This is compatible with the abc.ABCMeta metaclass
- Parameters:
callables (Callable[[Callable], Callable]) – function to wrap all callables with given class with
properties (Callable[[property], property]) – function to wrap all properties with given class with
selector_callables (Callable[[Callable], bool]) – additional criterion to be ran on given callable before deciding to wrap it. It must return True for wrapping to proceed.
selector_properties (Callable[[property], bool]) – additional criterion to be ran on given property before deciding to wrap it. It must return True for wrapping to proceed.
- satella.coding.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]