slipstream.utils

Slipstream utilities.

Attributes

AsyncCallable

Classes

Singleton	Maintain a single instance of a class.
PubSub	Singleton publish subscribe pattern class.

Functions

iscoroutinecallable(→ bool)	Check whether function is coroutine.
get_param_names(o)	Return function parameter names.

Module Contents

type slipstream.utils.AsyncCallable = Callable[..., Awaitable[Any]] | Callable[..., Any]

slipstream.utils.iscoroutinecallable(o: Any) → bool

Check whether function is coroutine.

slipstream.utils.get_param_names(o: Any)

Return function parameter names.

class slipstream.utils.Singleton

Bases: type

Maintain a single instance of a class.

class slipstream.utils.PubSub

Singleton publish subscribe pattern class.

subscribe(topic: str, listener: AsyncCallable) → None

Subscribe callable to topic.

unsubscribe(topic: str, listener: AsyncCallable) → None

Unsubscribe callable from topic.

publish(topic: str, *args: Any, **kwargs: Any) → None

Publish message to subscribers of topic.

async apublish(topic: str, *args: Any, **kwargs: Any) → None

Publish message to subscribers of topic.

async iter_topic(topic: str) → AsyncIterator[Any]

Asynchronously iterate over messages published to a topic.