Utils (objetto.utils)

General utilities.

Caller Module

Utilities to retrieve the caller’s module name.

objetto.utils.caller_module.get_caller_module(frames=0)

Get caller module name if possible.

Parameters: frames (int) – How many frames in the stack to go back relative to the caller.
Returns: Module name or None.
Return type: str or None

Custom Representations

Custom representation functions.

objetto.utils.custom_repr.custom_mapping_repr(mapping, prefix='{', template='{key}: {value}', separator=', ', suffix='}', sorting=False, sort_key=None, reverse=False, key_repr=<built-in function repr>, value_repr=<built-in function repr>)

Get custom representation of a mapping.

>>> from objetto.utils.custom_repr import custom_mapping_repr

>>> dct = {"a": 1, "b": 2}
>>> custom_mapping_repr(
...     dct, prefix="<", suffix=">", template="{key}={value}", sorting=True
... )
"<'a'=1, 'b'=2>"

Parameters

mapping (collections.abc.Mapping) – Mapping.
prefix (str) – Prefix.
template (str) – Item format template ({key} and {value}).
separator (str) – Separator.
suffix (str) – Suffix.
sorting (bool) – Whether to sort keys.
sort_key (function or None) – Sorting key.
reverse (bool) – Reverse sorting.
key_repr (function) – Key representation function.
value_repr (function) – Value representation function.

Returns

Custom representation.

Return type

str

objetto.utils.custom_repr.custom_iterable_repr(iterable, prefix='[', template='{value}', separator=', ', suffix=']', sorting=False, sort_key=None, reverse=False, value_repr=<built-in function repr>)

Get custom representation of an iterable.

>>> from objetto.utils.custom_repr import custom_iterable_repr

>>> tup = ("a", "b", "c", 1, 2, 3)
>>> custom_iterable_repr(tup, prefix="<", suffix=">", value_repr=str)
'<a, b, c, 1, 2, 3>'

Parameters

iterable (collections.abc.Iterable) – Iterable.
prefix (str) – Prefix.
template (str) – Item format template ({key} and {value}).
separator (str) – Separator.
suffix (str) – Suffix.
sorting (bool) – Whether to sort the iterable or not.
sort_key (function) – Sorting key.
reverse (bool) – Reverse sorting.
value_repr (function) – Value representation function.

Returns

Custom representation.

Return type

str

Dummy Context

Dummy context manager.

class objetto.utils.dummy_context.DummyContext

Dummy context manager.

>>> from objetto.utils.dummy_context import DummyContext

>>> with DummyContext():
...     pass

Factoring

Import and run factory functions/classes.

objetto.utils.factoring.get_factory_name(factory)

Get factory name without importing lazy paths.

>>> from objetto.utils.factoring import get_factory_name

>>> get_factory_name("module.submodule|Class.method")
'method'
>>> get_factory_name(int)
'int'

Parameters: factory (str or function or collections.abc.Callable or None) – Factory.
Returns: Factory name.
Return type: str

objetto.utils.factoring.format_factory(factory, module=None)

Check and format factory by adding a module path if applicable.

>>> from objetto.utils.factoring import format_factory

>>> format_factory("Class.method", module="module.submodule")
'module.submodule|Class.method'
>>> format_factory(int, module="module.submodule").__name__
'int'

Parameters

factory (str or function or collections.abc.Callable or None) – Factory.
module (str or None) – Module to prefix lazy paths without one.

Returns

Formatted factory.

Return type

str or function or collections.abc.Callable or None

Raises

ValueError – Invalid import path to factory.
TypeError – Did not provide valid factory.

objetto.utils.factoring.import_factory(factory)

Import factory from a path.

>>> from re import match
>>> from objetto.utils.factoring import import_factory

>>> import_factory("abc|abstractmethod")
<function abstractmethod at ...>
>>> import_factory(None)  # returns None
>>> import_factory(match)
<function match at ...>

Parameters: factory (str or function or collections.abc.Callable or None) – Lazy factory.
Returns: Imported factory.
Return type: function or collections.abc.Callable or None
Raises: TypeError – Invalid factory.

objetto.utils.factoring.run_factory(factory, args=None, kwargs=None)

Import and run factory.

>>> from objetto.utils.factoring import run_factory

>>> bool(run_factory("re|match", (r"^[a-z]+$", "abc")))
True

Parameters

factory (str or function or collections.abc.Callable or None) – Lazy factory.
args – Arguments to be passed to the factory function.
kwargs – Keyword arguments to be passed to the factory function.

Returns

Result from factory.

Lazy Import

Generate importable lazy paths and import from them.

objetto.utils.lazy_import.PRE_IMPORT_PATH_VALIDATION_REGEX = '^[\\w\\-\\|\\.]+$': Pre import path regex validation regex.

objetto.utils.lazy_import.PARTIAL_IMPORT_PATH_REGEX = '^((?:\\w+\\.{0,1})+)$': Partial lazy import path regex.

objetto.utils.lazy_import.RELATIVE_IMPORT_PATH_REGEX = '^((?:(?:\\.+)|(?:\\.*)(?:\\w+\\.{0,1})+))\\|((?:\\w+\\.{0,1})+)$': Relative lazy import path regex.

objetto.utils.lazy_import.IMPORT_PATH_REGEX = '^((?:\\w+\\.{0,1})+)\\|((?:\\w+\\.{0,1})+)$': Full lazy import path regex.

objetto.utils.lazy_import.import_path(path)

Import object from a full import path.

>>> from objetto.utils.lazy_import import import_path

>>> import_path("abc|abstractmethod")
<function abstractmethod at ...>

Parameters

path (str) – Import path.

Returns

Imported object.

Raises

ValueError – Invalid or empty path.
AttributeError – No object with the provided name.

objetto.utils.lazy_import.get_path(obj)

Get full import path to an object.

>>> from abc import abstractmethod
>>> from objetto.utils.lazy_import import get_path

>>> get_path(abstractmethod)
'abc|abstractmethod'

Parameters: obj – Object.
Returns: Import path.
Return type: str
Raises: ValueError – Can’t determine consistent import path.

objetto.utils.lazy_import.decorate_path(path, module=None)

Validate and decorate partial/relative import path with module if applicable.

>>> from objetto.utils.lazy_import import decorate_path

>>> decorate_path("abstractmethod", module="abc")
'abc|abstractmethod'

Parameters

path (str) – Import path (partial or full)
module (str or None) – Optional module path.

Returns

Full import path.

Return type

str

Raises

ValueError – Invalid path or no module provided.

List Operations

List operations helper functions.

objetto.utils.list_operations.resolve_index(length, index, clamp=False)

Resolve index to a positive number.

Parameters

length (int) – Length of the list.
index (int) – Input index.
clamp (bool) – Whether to clamp between zero and the length.

Returns

Resolved index.

Return type

int

Raises

IndexError – Index out of range.

objetto.utils.list_operations.resolve_continuous_slice(length, slc)

Resolve continuous slice according to length.

Parameters

length (int) – Length of the list.
slc (slice) – Continuous slice.

Returns

Index and stop.

Return type

tuple[int, int]

Raises

IndexError – Slice is noncontinuous.

objetto.utils.list_operations.pre_move(length, item, target_index)

Perform checks before moving values internally.

Parameters

length (int) – Length of the list.
item (int or slice) – Index/slice.
target_index (int) – Target index.

Returns

None or (index, stop, target index, post index).

Return type

None or tuple[int, int, int, int]

Qualified Name

Python-2 compatile way of finding a qualified name of a class/method.

objetto.utils.qualname.qualname(obj)

Find out the qualified name for a class or function.

Parameters: obj (function or type) – Function or class.
Returns: Qualified name.
Return type: str
Raises: AttributeError – Raised when couldn’t get qualified name.

Recursive Representation

Recursive-ready repr decorator.

@objetto.utils.recursive_repr.recursive_repr

Decorate a representation method/function and prevents infinite recursion.

>>> from objetto.utils.recursive_repr import recursive_repr

>>> class MyClass(object):
...
...     @recursive_repr
...     def __repr__(self):
...         return "MyClass<" + repr(self) + ">"
...
>>> my_obj = MyClass()
>>> repr(my_obj)
'MyClass<...>'

Parameters: func (function) – The ‘__repr__’ and/or ‘__str__’ method/function.
Returns: Decorated method function.
Return type: function

Re-Raise Context

Re-raise exceptions context manager.

class objetto.utils.reraise_context.ReraiseContext(exc_type=<class 'Exception'>, message=None)

Re-raise exceptions context manager for shorter tracebacks and custom messages.

>>> from objetto.utils.reraise_context import ReraiseContext

>>> def func_a():
...     func_b()
...
>>> def func_b():
...     func_c()
...
>>> def func_c():
...     raise ValueError("something is wrong")
...
>>> with ReraiseContext(ValueError, "oops"):
...     func_a()
Traceback (most recent call last):
ValueError: oops; something is wrong

Parameters

exc_type (type[Exception] or tuple[type[Exception]]) – Exception type(s) to catch.
message (str) – Optional additional custom message.

property exc_type

Exception type(s) to catch.

Return type: type[Exception] or tuple[type[Exception]]

property message

Optional additional custom message.

Return type: str or None

Simplify Exceptions

Decorator to simplify exception traceback when used with wrapped functions.

@objetto.utils.simplify_exceptions.simplify_exceptions

To be used with a wrapped function. This will simplify the exception traceback by chopping up two levels of it.

>>> from functools import wraps
>>> from objetto.utils.simplify_exceptions import simplify_exceptions

>>> def my_decorator(f):
...     @wraps(f)
...     @simplify_exceptions
...     def _wrapped(*args, **kwargs):
...         return f(*args, **kwargs)  # will be excluded from the traceback
...     return _wrapped
...
>>> @my_decorator
... def my_function():
...     raise ValueError("oops; something is wrong")
...
>>> my_function()
Traceback (most recent call last):
ValueError: oops; something is wrong

Parameters: func (function) – Function to be decorated.
Returns: Decorated function.
Return type: function

Storage

Immutable weak key/strong value storage and mutable evolver.

class objetto.utils.storage.AbstractStorage(*args, **kwds)

Abstract interface for storages.

abstract update(updates)

Update keys and values.

Parameters: updates (collections.abc.Mapping[collections.abc.Hashable, Any]) – Updates.
Returns: Updated abstract storage.
Return type: AbstractStorage

abstract query(key)

Query value for key.

Parameters: key (collections.abc.Hashable) – Key.
Returns: Value.
Raises: KeyError – Key is not in storage.

abstract to_dict()

Convert to dictionary.

Returns: Dictionary.
Return type: dict[collections.abc.Hashable, Any]

final class objetto.utils.storage.Storage(initial=None)

Immutable weak key/strong value storage.

Parameters: initial (collections.abc.Mapping[collections.abc.Hashable, Any]) – Initial values.

to_dict()

Convert to dictionary.

Returns: Dictionary.
Return type: dict[collections.abc.Hashable, Any]

update(updates)

Get new storage with update keys and values.

Parameters: updates (collections.abc.Mapping[collections.abc.Hashable, Any]) – Updates.
Returns: Updated storage.
Return type: Storage

query(key)

Query value for key.

Parameters: key (collections.abc.Hashable) – Key.
Returns: Value.
Raises: KeyError – Key is not in storage.

evolver()

Get evolver.

Returns: Evolver.
Return type: StorageEvolver

final class objetto.utils.storage.StorageEvolver(storage)

Mutable data storage evolver.

to_dict()

Convert to dictionary.

Returns: Dictionary.
Return type: dict[collections.abc.Hashable, Any]

update(updates)

Update keys and values in place.

Parameters: updates (collections.abc.Mapping[collections.abc.Hashable, Any]) – Updates.
Returns: Itself.
Return type: StorageEvolver

query(key)

Query value for key.

Parameters: key (collections.abc.Hashable) – Key.
Returns: Value.
Raises: KeyError – Key is not in storage.

persistent()

Get immutable storage with current updates.

Returns: Storage.
Return type: Storage

fork()

Fork into another evolver.

Returns: Forked evolver.
Return type: StorageEvolver

is_dirty()

Get whether has updates.

Returns: True if has updates.
Return type: bool

reset(): Reset updates.

commit(): Commit updates.

property updates

Updates.

Return type: pyrsistent.PMap[collections.abc.Hashable, Any]

Subject-Observer

Implementation of the Subject-Observer Pattern.

class objetto.utils.subject_observer.Subject

Sends payloads to observers.

__deepcopy__(memo=None)

Get a new subject, does not copy registered observers.

Returns: New subject.
Return type: objetto.utils.subject_observer.Subject

__copy__()

Get a new subject, does not copy registered observers.

Returns: New subject.
Return type: objetto.utils.subject_observer.Subject

__reduce__()

Reduce for pickling purposes.

Returns: Subject class, no arguments.
Return type: tuple[type[objetto.utils.subject_observer.Subject], tuple]

wait(token)

Wait for the token’s observer to receive the payload before continuing.

Parameters

token (objetto.utils.subject_observer.ObserverToken) – Observer token.

Raises

ValueError – Token does not belong to this subject.
RuntimeError – Token cycle detected.
RuntimeError – Can’t wait for failed observer.

send(*payload)

Send payload to all observers.

Parameters: payload – Payload.
Returns: Exception infos (for exceptions raised during observers’ responses).
Return type: tuple[objetto.utils.subject_observer.ObserverExceptionInfo]
Raises: RuntimeError – Already sending.

register_observer(observer)

Register an observer and get its token.

Parameters: observer (objetto.utils.subject_observer.Observer) – Observer.
Returns: Observer token.
Return type: objetto.utils.subject_observer.ObserverToken

deregister_observer(observer)

De-register an observer.

Parameters: observer (objetto.utils.subject_observer.Observer) – Observer.

get_token(observer)

Get token for already registered observer.

Parameters: observer (objetto.utils.subject_observer.Observer) – Observer.
Returns: Observer token.
Return type: objetto.utils.subject_observer.ObserverToken
Raises: ValueError – Observer is not registered.

class objetto.utils.subject_observer.Observer

Observes payloads sent from a subject.

>>> from objetto.utils.subject_observer import Subject, Observer

>>> class MyObserver(Observer):
...
...     def __observe__(self, *payload):
...         print("received payload {}".format(payload))
...
>>> subject = Subject()
>>> observer = MyObserver()
>>> token = observer.start_observing(subject)
>>> exception_infos = subject.send(1, 2, 3)
received payload (1, 2, 3)

abstract __observe__(*payload)

Receive a payload sent from a subject and react to it.

Parameters: payload – Payload.
Raises: NotImplementedError – Abstract method not implemented.

start_observing(subject)

Start observing a subject.

Parameters: subject (objetto.utils.subject_observer.Subject) – Subject.
Returns: Observer token.
Return type: objetto.utils.subject_observer.ObserverToken

stop_observing(subject)

Stop observing a subject.

Parameters: subject (objetto.utils.subject_observer.Subject) – Subject.

class objetto.utils.subject_observer.ObserverToken(subject, observer)

Allows control over observers’ order/priority.

Note

This class can’t be instantiated directly. A token should be retrieved when an observer starts observing a subject or by using the Subject.get_token() method.

>>> from objetto.utils.subject_observer import Subject, Observer

>>> class MyObserver(Observer):
...     def __init__(self, name, actions, dependency=None):
...         self.name = name
...         self.actions = actions
...         self.dependency = dependency
...
...     def __observe__(self, *payload):
...         if self.dependency:
...             self.dependency.wait()
...         action = "{} received payload {}".format(self.name, payload)
...         self.actions.append(action)
...
>>> subject = Subject()
>>> executed_actions = []

>>> observer_a = MyObserver("A", executed_actions)
>>> token_a = observer_a.start_observing(subject)

>>> observer_b = MyObserver("B", executed_actions, dependency=token_a)
>>> token_b = observer_b.start_observing(subject)

>>> exception_infos = subject.send(1, 2, 3)
>>> executed_actions  # note how A will always receive the payload before B
['A received payload (1, 2, 3)', 'B received payload (1, 2, 3)']

wait(): Wait for the associated observer to finish observing before continuing.

property observer

Observer.

Return type: objetto.utils.subject_observer.Observer or None

class objetto.utils.subject_observer.ObserverExceptionInfo(observer, payload, exception_type, exception, traceback)

Describes an exception raised by an observer receiving a payload.

Parameters

observer (objetto.utils.subject_observer.Observer) – Observer.
payload (tuple) – Payload.
exception_type (type[BaseException] or None) – Exception type.
exception (BaseException or None) – Exception.
traceback (types.TracebackType or None) – Traceback.

Type Checking

Runtime type checking with support for lazy import paths.

objetto.utils.type_checking.format_types(types, module=None)

Check and format types by adding a module path.

>>> from objetto.utils.type_checking import format_types, get_type_names

>>> get_type_names(format_types((int, "chain"), module="itertools"))
('int', 'chain')
>>> get_type_names(format_types(float))
('float',)
>>> get_type_names(format_types("itertools|chain"))
('chain',)

Parameters

types (str or type or None or tuple[str or type or None]) – Types.
module (str or None) – Module to prefix lazy paths without one.

Returns

Formatted types.

Return type

tuple[str or type or None]

Raises

ValueError – Invalid type path/no module provided.
TypeError – Did not provide valid types.

objetto.utils.type_checking.get_type_names(types)

Get type names without importing lazy paths.

>>> from objetto.utils.type_checking import get_type_names

>>> get_type_names((int, "itertools|chain"))
('int', 'chain')

Parameters: types (str or type or None or tuple[str or type or None]) – Types.
Returns: Type names.
Return type: tuple[str]

objetto.utils.type_checking.flatten_types(types)

Flatten types into a tuple.

>>> from objetto.utils.type_checking import flatten_types, get_type_names

>>> get_type_names(flatten_types((int, ("itertools|chain", float, (str,)))))
('int', 'chain', 'float', 'str')
>>> get_type_names(flatten_types(int))
('int',)

Parameters: types (str or type or None or tuple[str or type or None]) – Types.
Returns: Flattened types.
Return type: tuple[str or type]
Raises: TypeError – Invalid types.

objetto.utils.type_checking.import_types(types)

Import types from lazy import paths.

>>> from objetto.utils.type_checking import import_types, get_type_names

>>> get_type_names(import_types("itertools|chain"))
('chain',)
>>> get_type_names(import_types(("itertools|chain", "itertools|compress")))
('chain', 'compress')

Parameters: types (str or type or None or tuple[str or type or None]) – Types.
Returns: Imported types.
Return type: tuple[type]

objetto.utils.type_checking.is_instance(obj, types, subtypes=True)

Get whether object is an instance of any of the provided types.

>>> from itertools import chain
>>> from objetto.utils.type_checking import is_instance

>>> class SubChain(chain):
...     pass
...
>>> is_instance(3, int)
True
>>> is_instance(3, (chain, int))
True
>>> is_instance(3, ())
False
>>> is_instance(SubChain(), "itertools|chain")
True
>>> is_instance(chain(), "itertools|chain", subtypes=False)
True
>>> is_instance(SubChain(), "itertools|chain", subtypes=False)
False

Parameters

obj (object) – Object.
types (str or type or None or tuple[str or type or None]) – Types.
subtypes (bool) – Whether to accept subtypes.

Returns

True if it is an instance.

Return type

bool

objetto.utils.type_checking.is_subclass(cls, types, subtypes=True)

Get whether class is a subclass of any of the provided types.

>>> from itertools import chain
>>> from objetto.utils.type_checking import is_subclass

>>> class SubChain(chain):
...     pass
...
>>> is_subclass(int, int)
True
>>> is_subclass(int, (chain, int))
True
>>> is_subclass(int, ())
False
>>> is_subclass(SubChain, "itertools|chain")
True
>>> is_subclass(chain, "itertools|chain", subtypes=False)
True
>>> is_subclass(SubChain, "itertools|chain", subtypes=False)
False

Parameters

cls (type) – Class.
types (str or type or None or tuple[str or type or None]) – Types.
subtypes (bool) – Whether to accept subtypes.

Returns

True if it is a subclass.

Return type

bool

Raises

TypeError – Did not provide a class.

objetto.utils.type_checking.assert_is_instance(obj, types, subtypes=True)

Assert object is an instance of any of the provided types.

>>> from itertools import chain
>>> from objetto.utils.type_checking import assert_is_instance

>>> class SubChain(chain):
...     pass
...
>>> assert_is_instance(3, int)
>>> assert_is_instance(3, (chain, int))
>>> assert_is_instance(3, ())
Traceback (most recent call last):
ValueError: no types were provided to perform assertion
>>> assert_is_instance(3, "itertools|chain")
Traceback (most recent call last):
TypeError: got 'int' object, expected instance of 'chain' or any of its subclasses
>>> assert_is_instance(chain(), "itertools|chain", subtypes=False)
>>> assert_is_instance(SubChain(), "itertools|chain", subtypes=False)
Traceback (most recent call last):
TypeError: got 'SubChain' object, expected instance of 'chain' (instances of subclasses are not accepted)

Parameters

obj (object) – Object.
types (str or type or None or tuple[str or type or None]) – Types.
subtypes (bool) – Whether to accept subtypes.

Raises

ValueError – No types were provided.
TypeError – Object is not an instance of provided types.

objetto.utils.type_checking.assert_is_subclass(cls, types, subtypes=True)

Assert a class is a subclass of any of the provided types.

>>> from itertools import chain
>>> from objetto.utils.type_checking import assert_is_subclass

>>> class SubChain(chain):
...     pass
...
>>> assert_is_subclass(int, int)
>>> assert_is_subclass(int, (chain, int))
>>> assert_is_subclass(int, ())
Traceback (most recent call last):
ValueError: no types were provided to perform assertion
>>> assert_is_subclass(int, "itertools|chain")
Traceback (most recent call last):
TypeError: got 'int', expected class 'chain' or any of its subclasses
>>> assert_is_subclass(chain, "itertools|chain", subtypes=False)
>>> assert_is_subclass(SubChain, "itertools|chain", subtypes=False)
Traceback (most recent call last):
TypeError: got 'SubChain', expected class 'chain' (subclasses are not accepted)

Parameters

cls (type) – Class.
types (str or type or None or tuple[str or type or None]) – Types.
subtypes (bool) – Whether to accept subtypes.

Raises

ValueError – No types were provided.
TypeError – Class is not a subclass of provided types.

objetto.utils.type_checking.assert_is_callable(value)

Assert a value is callable.

>>> from objetto.utils.type_checking import assert_is_subclass

>>> assert_is_callable(int)
>>> assert_is_callable(lambda: None)
>>> assert_is_callable(3)
Traceback (most recent call last):
TypeError: got non-callable 'int' object, expected a callable

Parameters: value – Value.
Raises: TypeError – Value is not a match.

Weak Reference

Weak reference type.

class objetto.utils.weak_reference.WeakReference(obj=None)

Weak reference object that supports pickling.

Inherits from:

typing.Generic

>>> from objetto.utils.weak_reference import WeakReference

>>> class MyClass(object):
...     pass
...
>>> strong = MyClass()
>>> weak = WeakReference(strong)
>>> weak() is strong
True
>>> del strong
>>> weak() is None
True

Parameters: obj (object) – Object to reference.

__hash__()

Get hash.

Returns: Hash.
Return type: int

__eq__(other)

Compare for equality.

Parameters: other – Other.
Returns: True if equal.
Return type: bool or NotImplemented

__ne__(other)

Compare for inequality.

Parameters: other – Other.
Returns: True if not equal.
Return type: bool or NotImplemented

__repr__()

Get representation.

Returns: Representation.
Return type: str

__str__()

Get string representation.

Returns: String representation.
Return type: str

__call__()

Get strong reference to the object or None if no longer alive.

Returns: Strong reference or None.
Return type: object or None

__reduce__()

Reduce for pickling.

Returns: Class and strong reference to object.
Return type: tuple[type[objetto.utils.weak_reference.WeakReference], tuple[object]]