core.util¶

immp.core.util.resolve_import(path)¶

Take the qualified name of a Python class, and return the physical class object.

Parameters:: path (str) – Dotted Python class name, e.g. <module path>.<class name>.
Returns:: Class object imported from module.
Return type:: type

immp.core.util.pretty_str(cls)¶: Class decorator to provide a default __str__() based on the contents of __dict__.

immp.core.util.escape(raw, *chars)¶

Prefix special characters with backslashes, suitable for encoding in a larger string delimiting those characters.

Parameters:

raw (str) – Unsafe input string.
chars (str list) – Control characters to be escaped.

Returns:

Escaped input string.

Return type:

str

immp.core.util.unescape(raw, *chars)¶

Inverse of escape(), remove backslashes escaping special characters.

Parameters:

raw (str) – Escaped input string.
chars (str list) – Control characters to be unescaped.

Returns:

Raw unescaped string.

Return type:

str

class immp.core.util.Watchable(watch)¶

Bases: object

Container mixin to trigger a callback when its contents are changed.

classmethod unwrap(obj)¶

Recursively replace Watchable subclasses with their native equivalents.

Parameters:: obj – Container type, or any value.
Returns:: Unwrapped container, or the original value.

classmethod watch(*methods)¶

Class decorator for mixin users, to wrap methods that modify the underlying container, and should therefore trigger the callback when invoked.

Parameters:: methods (str list) – List of method names to individually wrap.

class immp.core.util.WatchedDict(watch, initial, **kwargs)¶

Bases: dict, Watchable

Watchable-enabled dict subclass. Lists or dictionaries added as items in this container will be wrapped automatically.

update([E, ]**F) → None. Update D from dict/iterable E and F.¶: If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

clear() → None. Remove all items from D.¶

pop(k[, d]) → v, remove specified key and return the corresponding value.¶: If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()¶

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

setdefault(key, default=None, /)¶

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

class immp.core.util.WatchedList(watch, initial)¶

Bases: list, Watchable

Watchable-enabled list subclass. Lists or dictionaries added as items in this container will be wrapped automatically.

insert(index, value)¶: Insert object before index.

append(value)¶: Append object to the end of the list.

extend(other)¶: Extend list by appending elements from the iterable.

clear()¶: Remove all items from list.

pop(index=-1, /)¶

Remove and return item at index (default last).

Raises IndexError if list is empty or index is out of range.

remove(value, /)¶

Remove first occurrence of value.

Raises ValueError if the value is not present.

reverse()¶: Reverse IN PLACE.

sort(*, key=None, reverse=False)¶

Sort the list in ascending order and return None.

The sort is in-place (i.e. the list itself is modified) and stable (i.e. the order of two equal elements is maintained).

If a key function is given, apply it once to each list item and sort them, ascending or descending, according to their function values.

The reverse flag can be set to sort in descending order.

class immp.core.util.ConfigProperty(cls=None, key=None)¶

Bases: object

Data descriptor to present config from Openable instances using the actual objects stored in a Host.

class immp.core.util.IDGen¶

Bases: object

Generator of generic timestamp-based identifiers.

IDs are guaranteed unique for the lifetime of the application – two successive calls will yield two different identifiers.

class immp.core.util.LocalFilter(name='')¶

Bases: Filter

Logging filter that restricts capture to loggers within the immp namespace.

Deprecated since version 0.10.0: Pure logging config offers a cleaner solution to using this filter::

root:
level: WARNING

loggers:

immp:
level: DEBUG

filter(record)¶

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

class immp.core.util.OpenState(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)¶

Bases: Enum

Readiness status for instances of Openable.

disabled¶: Not currently in use, and won’t be started by the host.

inactive¶: Hasn’t been started yet.

starting¶: Currently starting up (during Openable.start()).

active¶: Currently running.

stopping¶: Currently closing down (during Openable.stop()).

failed¶: Exception occurred during starting or stopping phases.

class immp.core.util.Configurable(name, config, host)¶

Bases: object

Superclass for objects managed by a Host and created using configuration.

schema¶

Structure of the config expected by this configurable. If not customised by the subclass, it defaults to dict (that is, any dict structure is valid).

It may also be set to None, to declare that no configuration is accepted.

Type:: .Schema

name¶

User-provided, unique name of the hook, used for config references.

Type:: str

config¶

Reference to the user-provided configuration. Assigning to this field will validate the data against the class schema, which may raise exceptions on failure as defined by Schema.validate().

Type:: dict

host¶

Controlling host instance, providing access to plugs.

Type:: .Host

class immp.core.util.Openable¶

Bases: object

Abstract class to provide open and close hooks. Subclasses should implement start() and stop(), whilst users should make use of open() and close().

state¶

Current status of this resource.

Type:: .OpenState

async open()¶: Open this resource ready for use. Does nothing if already open, but raises RuntimeError if currently changing state.

async start()¶

Perform any underlying operations needed to ready this resource for use, such as opening connections to an external network or API.

If using an event-driven framework that yields and runs in the background, you should use a signal of some form (e.g. asyncio.Condition) to block this method until the framework is ready for use.

async close()¶: Close this resource after it’s used. Does nothing if already closed, but raises RuntimeError if currently changing state.

async stop()¶

Perform any underlying operations needed to stop using this resource and tidy up, such as terminating open network connections.

Like with start(), this should block as needed to wait for other frameworks – this method should return only when ready to be started again.

disable()¶: Prevent this openable from being run by the host.

enable()¶: Restore normal operation of this openable.

class immp.core.util.HTTPOpenable¶

Bases: Openable

Template openable including a aiohttp.ClientSession instance for networking.

session¶

Managed session object.

Type:: aiohttp.ClientSession

async start()¶

Perform any underlying operations needed to ready this resource for use, such as opening connections to an external network or API.

If using an event-driven framework that yields and runs in the background, you should use a signal of some form (e.g. asyncio.Condition) to block this method until the framework is ready for use.

async stop()¶

Perform any underlying operations needed to stop using this resource and tidy up, such as terminating open network connections.

Like with start(), this should block as needed to wait for other frameworks – this method should return only when ready to be started again.