slipstream.utils

Slipstream utilities.

Attributes

T
P
AsyncCallable
Pipe

Classes

Signal	Signals can be exchanged with streams.
Singleton	Maintain a single instance of a class.
PubSub	Singleton publish subscribe pattern class.
AsyncSynchronizedGenerator	Async generator that synchronizes values across copies.

Functions

awaitable(→ Any)	Convert into awaitable.
iscoroutinecallable(→ bool)	Check whether object is coroutine.
get_param_names(→ tuple[str, Ellipsis])	Return function parameter names.

Module Contents

slipstream.utils.T

slipstream.utils.P

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

type slipstream.utils.Pipe = Callable[[AsyncIterable[Any]], AsyncIterable[Any]]

class slipstream.utils.Signal

Bases: enum.Enum

Signals can be exchanged with streams.

SENTINEL represents an absent yield value PAUSE represents the signal to pause stream RESUME represents the signal to resume stream

SENTINEL = 0

PAUSE = 1

RESUME = 2

STOP = 3

async slipstream.utils.awaitable(x: Any) → Any

Convert into awaitable.

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

Check whether object is coroutine.

slipstream.utils.get_param_names(o: Any) → tuple[str, Ellipsis]

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) → collections.abc.AsyncIterator[Any]

Asynchronously iterate over messages published to a topic.

class slipstream.utils.AsyncSynchronizedGenerator(gen: collections.abc.AsyncIterable[Any])

Async generator that synchronizes values across copies.

property value: Any

Get current value the generator is holding.

copy() → _GeneratorCopy

Create a synchronized copy of this generator.