ubii.framework.util.functools module

ubii.framework.util.functools.similar(choices: Sequence, item: Any, cutoff=0.70)

Use this e.g. if you think your users can’t type <:

Example

>>> >>> from ubii.framework import util
>>> choices = ["Foo", "Bar", "Foobar", "Thing"]
>>> util.similar(choices, "Thong")
['Thing']
>>> util.similar(choices, "foo")
[]
>>> util.similar(choices, "foo", cutoff=0.5)
['Foo']
>>> util.similar(choices, "foo", cutoff=0.4)
['Foo', 'Foobar']

Parameters:

• choices – sequence of items

• item – something not in choices but possibly ‘similar’

• cutoff – threshold for similarity measure

Returns:

elements from choices with \(similarity(choice, item) > cutoff\)

See also

difflib.SequenceMatcher – used internally to compute similarity score

class ubii.framework.util.functools.hook(func: T_Callable, decorators=None)

Bases: Generic[T_Callable]

This decorator gives the decorated callable the ability to register decorators, i.e. define a consistent API to apply decorators to the decorated callable.

Notes

• Decorators need not to be unique, registering the same decorator multiple times will apply it multiple times.

• Decorators are applied at the first call to the hook, i.e. the first call could take longer.

• hook.func is the unmodified callable passed during initialization. If necessary one can clear the cached ‘applied’ version of the callable with hook.cache_clear()

Warning

Decorator order depends on order of registration. The last registered decorator is applied last. We have \(hook = d_n \circ d_{n-1} \circ ... \circ d_1 \circ d_0 \circ f\) where \(0\) to \(n\) are the indices of the decorator in hook.decorators

Example

You have some callable which is predestined to be slightly altered later. Instead of monkey patching it later, you can preemptively define that callable to be ‘alterable’ by converting it into a hook.

>>> from ubii.framework import util
>>> value_factory = iter([1, 2, 70, 3, 4, 5]).__next__
>>> hook = util.hook(value_factory)
>>> hook()
1
>>> hook()
2
>>> hook()
70
>>> def decorator(func):
...     def inner():
...             return -1 * func()
...     return inner
...
>>> hook.register_decorator(decorator)
>>> hook()
-3
>>> hook()
-4

__init__(func: T_Callable, decorators=None)

Create a hook from func

Parameters:

• func – a callable which should be more easily decorate-able

• decorators – initial decorators – optional

func

Reference to unmodified callable

decorators(instance: object | None = None)

List of registered decorators

Parameters:

instance – return decorators for this specific instance – optional

register_decorator(decorator, instance: object | None = None) → None

Add decorator to internal list of decorators and re-apply all decorators at next call

Parameters:

• decorator – some callable

• instance – only register decorator for this specific instance – optional

cache_clear()

Apply decorators again at next access

class ubii.framework.util.functools.registry(key: Callable[[T], S], fn: Callable[[...], T])

Bases: Generic[S, T]

Decorator to register every call to another callable

Example

Let’s say we have another decorator that does some simple task

>>> def my_decorator(func):
...     def inner(*args):
...         print("Foo")
...         return func(*args)
...     return inner

We want to create a new decorator that does the same but keeps track of every decorated function

>>> from ubii.framework import util
>>> my_decorator_registry = util.registry(key=lambda func: func.__name__, fn=my_decorator)
>>> @my_decorator_registry
... def test_function(foo: str):
...     print(foo)
...
>>> my_decorator_registry.registry
{'test_function': <function test_function at (...)>}
>>> test_function("Bar")
Foo
Bar

__init__(key: Callable[[T], S], fn: Callable[[...], T])

This callable wraps the callable passed as fn, but caches results in registry with key(result) as key.

Parameters:

• key – computes some hashable unique value for possible results of wrapped callable

• fn – if this callable returns non-unique results, old cached values are technically overwritten since the key returns the same value for the same input by definition

key

computes keys for results of fn to cache them inside registry

fn

wrapped callable

property registry: Dict[S, T]

Mapping \(key \rightarrow result\) from computed key(result) of all calls

ubii.framework.util.functools.exc_handler_decorator(handler: Callable[[ExcInfo], Awaitable[None | bool] | None | bool])

This callable takes an ‘exception handler’ i.e. a callable that processes results of sys.exc_info() and converts it to a decorator that can be itself applied to async callables to handle their exceptions.

Example

>>> def handler(*exc_info):
...     if exc_info:
...             print(f"Got exception {exc_info[1]}")

>>> from ubii.framework import util
>>> handler_deco = util.exc_handler_decorator(handler)
>>> @handler_deco
... async def foo(value):
...     print(5 // value)

>>> async def main():
...     for value in [0,1,2,0,-2]:
...             await foo(value)

>>> import asyncio
>>> asyncio.run(main())
Got exception integer division or modulo by zero
5
2
Got exception integer division or modulo by zero
-3

Parameters:

handler – an exception handler callable

Returns:

decorator to catch exceptions from async methods

ubii.framework.util.functools.log_call(logger: logging.Logger | None = None)

Parameters:

logger – calls are logged as logging.Logger.debug(), if not passed use logger with module name

Returns:

A decorator to log calls to decorated callable to logger

class ubii.framework.util.functools.ProtoRegistry(*args, **kwargs)

Bases: ProtoMeta, RegistryMeta

Instances for types that have this metaclass are registered, and can be serialized / deserialized according to their protobuf specifications.

See also

RegistryMeta – the ProtoRegistry simply adds the serialization

to the mechanisms for registration of instances inherited from here

save_specs(path)

Serialize all registered Protocol Buffer Wrapper objects and pickle the results to path

update_specs(path) → None

Loads specs from pickled file, updates all registered instances according to their specification from the pickled messages.

Parameters:

path – load binary pickle file from here

class ubii.framework.util.functools.function_chain(*funcs)

Bases: object

Generates a callable that calls multiple functions in a defined order with same arguments

Example

>>> def foo(value):
...     print(f"foo: {value}")
>>> def bar(value):
...     print(f"bar: {value}")
>>> from ubii.framework import util
>>> chain = util.function_chain(foo, bar)
>>> chain(1)
foo: 1
bar: 1

See also

compose – if you want to compose functions instead of passing the same arguments to each

async_compose – if you want to compose coroutines

funcs

Tuple of functions that need to be called

classmethod reverse(*funcs)

class ubii.framework.util.functools.compose(*funcs)

Bases: object

Generates a callable that is the composition of other callables. Callables are called in the order in which they are passed to compose i.e. \(compose(f, g) = g \circ f\)

Example

>>> from ubii.framework import util
>>> def foo(value):
...     return value + 1
...
>>> def bar(value):
...     print(value)
...
>>> foobar = util.compose(foo, bar)  # foobar(value) = bar(foo(value))
>>> foobar(1)
2

See also

function_chain – if you want to pass the same argument to every function instead of composing

async_compose – if you want to compose coroutines

funcs

Tuple of original callables

class ubii.framework.util.functools.make_dict(key: Callable[[Any], S], value: Callable[[Any], T], filter_none=False)

Bases: Generic[S, T]

This callable creates dictionaries using a key-function and a value-function to extract information from the values produced by some iterable.

Example

>>> from ubii.framework import util
>>> from collections import namedtuple
>>> from random import sample
>>> Foo = namedtuple('Foo', ['id', 'value'])
>>> def random_foo(k):
...     return [Foo(*v) for v in zip(range(k), sample(range(k), k))]
...
>>> id_to_value = util.make_dict(key=lambda f: f.id, value=lambda f: f.value)
>>> some_foos = random_foo(5)
>>> some_foos
[Foo(id=0, value=2), Foo(id=1, value=0), Foo(id=2, value=1), Foo(id=3, value=3), Foo(id=4, value=4)]
>>> id_to_value(some_foos)
{0: 2, 1: 0, 2: 1, 3: 3, 4: 4}
>>> more_foos = random_foo(3)
>>> more_foos
[Foo(id=0, value=0), Foo(id=1, value=2), Foo(id=2, value=1)]
>>> id_to_value(more_foos)
{0: 0, 1: 2, 2: 1}

class ubii.framework.util.functools.async_compose(*fns)

Bases: object

Like compose for coroutines. Coroutines are awaited in the order in which they are passed to async_compose i.e. composed = async_compose(f, g) is equivalent to async def composed(*args): return await g(await f(*args))

Example

>>> from ubii.framework import util
>>> import asyncio, random
>>> async def simulate_IO(value):
...     await asyncio.sleep(random.random())
...     return value
...
>>> async def make_values(k):
...     for n in range(k):
...             yield await simulate_IO(n)
...
>>> async def process(values):
...     return [value async for value in values if value % 2 == 0]
...
>>> async def result(values):
...     for value in values:
...             print(await simulate_IO(value))
...
>>> processed = util.compose(make_values, process)  # make_values does not return a coroutine -> compose
>>> composed = util.async_compose(processed, result) # processed returns a coroutine -> async_compose
>>> asyncio.run(composed(5))
0
2
4

See also

function_chain – if you want to pass the same argument to every function instead of composing

compose – if you want to compose normal callables instead of coroutines

class composed(f: Callable[[...], Awaitable], g: Callable[[...], Awaitable])

Bases: object

class ubii.framework.util.functools.enrich(meta, func: Callable)

Bases: Callable[[…], enrich.result]

Enriches the results of a callable with some meta information.

Useful if you deal with callables which you can’t or don’t want to change, to avoid leaking dependencies – e.g. the part of the codebase that “knows” the meta information is not the same that defines the callables, and it’s undesirable to change that.

Works with normal callables and also coroutine functions – if async coroutine functions / methods are used, the resulting callable will await the result first, before it adds the meta information and returns it to the caller.

Example

Consider a mapping of callables:

>>> import functools
>>> make_print = functools.partial(functools.partial, print)
>>> call_mapping = {value: make_print(value) for value in ['foo', 'bar', 'foobar']}
>>> call_mapping['foo']()
foo

Pretend that make_print is defined somewhere else in your code, maybe even in someone else’s code. You know that it’s a factory for callables that print one value, and the code where you define the call mapping also knows which value each callable in the mapping will print.

You don’t want to pass this meta information around in your code (let’s say as additional arguments for your calls), just because the implementation of make_print does not behave the way you want – i.e. it does not also return the value that gets printed.

So instead, you could define your mapping like this:

>>> from ubii.framework import util
>>> call_mapping = {v: util.enrich(v, make_print(v)) for v in ['foo', 'bar', 'foobar']}
>>> result = call_mapping['foo']()
foo
>>> result
result(value=None, meta='foo')

Now your code can simply deal with the “fixed” callables.

class result(value, meta)

Bases: tuple

Named tuple for access of original return value and meta information

meta

Alias for field number 1

value

Alias for field number 0

__init__(meta, func: Callable)

Add meta information to results of callable

Parameters:

• meta – Something that each call to this callable should return in the meta field

• func – Simple callable or coroutine function. Result will be available as value field

class ubii.framework.util.functools.calc_delta(get_value)

Bases: object

Calculate difference between values produced by a factory function

Example

>>> from ubii.framework import util
>>> value_factory = iter([1, 2, 70, 3, 4, 5]).__next__
>>> delta = util.calc_delta(value_factory)
>>> delta()
1
>>> delta()
68
>>> delta()
-67
>>> delta.value
3

class ubii.framework.util.functools.dunder

Bases: object

the following decorators can be used to create dunder methods on classes.

Note

Why not use attrs?

• because the ubii framework should be usable by people that don’t have a lot of experience in python, so it uses the standard library whenever possible -> dataclasses instead of attrs

• dataclasses are not useful for complex classes, nonetheless even a very complex class might have very easy dunder methods

classmethod all(*attrs)

A decorator that applies all decorators from dunder

Parameters:

*attrs – the names of attributes that should be part of the dunder methods

Returns:

a class decorator

classmethod repr(*attrs, patch_str_for_builtins: bool = True)

Creates a __repr__ method on the class, that is basically

def __repr__(self):
    info = {attr: getattr(self, attr, None) for attr in attrs}
    return f"{self.__class__.__name__}({', '.join('{}={}'.format(k, v) for k,v in info.items())})"

Parameters:

• *attrs – the names of attributes that should be part of the __repr__

• patch_str_for_builtins – if True, types that inherit directly from builtins will receive a __str__ method as well (so set this to false if you implement __str__ yourself) – optional

Returns:

a class decorator

classmethod hash(*attrs)

Creates a __hash__ method on the class, that is basically

def __hash__(self):
    return hash((self.__class__, ) + tuple(map(lambda attr: getattr(self, attr, None), sorted(attrs))))

Parameters:

*attrs – the names of attributes that should be part of the __hash__

Returns:

a class decorator

ubii.framework.util.functools.document_decorator(decorator)

class ubii.framework.util.functools.append_doc(cb)

Bases: object

Helper to append information to docstrings of callables