satella.coding package

Subpackages

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:

DictDeleter

iter_mode
keys()
Return type:

DictDeleter

keys_to_delete
values()
Return type:

DictDeleter

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

static get()

Return a local context for this thread

Return type:

Context

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 closed

  • kwargs – 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:

RunActionAfterGeneratorCompletes

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.

Deprecated since version 2.24.2: Use Optional(clbl)(*args, **kwargs) instead

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:

TypeSignature

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

static get()

Return a local context for this thread

Return type:

Context

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:

DictDeleter

iter_mode
keys()
Return type:

DictDeleter

keys_to_delete
values()
Return type:

DictDeleter

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 closed

  • kwargs – 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:

TypeSignature

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:

RunActionAfterGeneratorCompletes

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]