asyncutils.events

Classes that extend the functionality of Event with the interface it specifies, without inheriting from it.

Not at all related to the asyncio.events submodule, which manages the event loop, despite the common name.

Classes

EventWithValue	An event class that can store a value and maintains a history of past values.
SingleWaiterEventWithValue	Essentially wraps a future in the event interface.

Module Contents

class asyncutils.events.EventWithValue[T: ty_extensions.Not[None]](*, maxhist: int | None = ...)

Bases: asyncutils.mixins.EventMixin[T]

An event class that can store a value and maintains a history of past values.

A maximum of maxhist entries, which defaults to context.EVENT_WITH_VALUE_DEFAULT_MAX_HIST, of past results, are stored.

clear() → None

Unset the result of the event.

get(default: T = ...) → T

Get the result of the event immediately if set, otherwise returning default if passed or throw RuntimeError.

is_set() → bool

Whether the result is currently set.

recent_history(duration: float | None = ...) → collections.abc.Generator[tuple[float, T]]

Yield recent history entries in order; what qualifies as recent depends on duration, defaulting to context.EVENT_WITH_VALUE_DEFAULT_RECENT.

remove_done_waiters() → None

Should be run periodically to cleanup the internal queue of waiters, removing those having already completed.

set(value: None, *, strict: Literal[False]) → None

set(value: T, *, strict: bool = ...) → None

Set the result of the event and wake up waiters. If strict is True, throws an error when the value is None, since it is more idiomatic to call clear() instead.

set_once(value: T) → None

Set the result to value, and then immediately revert it to the original. Waiters are triggered twice.

async wait_for_next(timeout: float | None = ...) → T

Wait for the next result of the event to be set.

async wait_for_transition(old: T, new: T, timeout: float | None = ..., *, force_transition: bool = ..., legacy: bool = ...) → bool

Wait until the value is set to old, and then new, in that order.

timeout defaults to context.EVENT_WITH_VALUE_DEFAULT_TIMEOUT

On timeout, if force_transition is True, cause the transition to happen manually.

Return whether the transition occurred naturally.

Changed in version 0.9.6: Fixed a bug where overlapping potential transitions are not considered. The old behaviour can be achieved by passing legacy=True.

async wait_for_transition_unordered(

a: T,

b: T,

timeout: float | None = ...,

*,

force_transition: bool = ...,

legacy: bool = ...,

) → bool

Wait until either a transitions to b or b transitions to a, with the preference being for the former.

property history: list[tuple[float, T]]

The past results of the event as a list of tuples (timestamp, value).

property history_asdict: dict[float, T]

Above, but as a dictionary.

class asyncutils.events.SingleWaiterEventWithValue[T]

Bases: asyncutils.mixins.EventMixin[T]

Essentially wraps a future in the event interface.

clear() → None

Unset the result of the event.

get(default: T = ...) → T

Get the result of the event immediately if set, otherwise returning default if passed or throw RuntimeError.

is_set() → bool

Whether the result is currently set.

set(value: T) → None

Set the result of the event to value, awakening the sole waiter.

async wait(timeout: float | None = ..., *, strict: bool = ...) → T

async wait_for_next(timeout: float | None = ..., *, strict: bool = ...) → T

Wait for the next result of the event to be set.