asphalt.core.event

class asphalt.core.event.Event(source, topic, time=None)

The base class for all events.

Parameters:

  • source (Any) – the object where this event originated from

  • topic (str) – the event topic

  • time (float | None) – the time the event occurred

Variables:

  • source – the object where this event originated from

  • topic (str) – the topic

  • time (float) – event creation time as seconds from the UNIX epoch

property utc_timestamp: datetime

Return a timezone aware datetime corresponding to the time variable, using the UTC timezone.

class asphalt.core.event.Signal(event_class, *, source=None, topic=None)

Declaration of a signal that can be used to dispatch events.

This is a descriptor that returns itself on class level attribute access and a bound version of itself on instance level access. Connecting listeners and dispatching events only works with these bound instances.

Each signal must be assigned to a class attribute, but only once. The Signal will not function correctly if the same Signal instance is assigned to multiple attributes.

Parameters:

event_class – an event class

connect(callback)

Connect a callback to this signal.

Each callable can only be connected once. Duplicate registrations are ignored.

If you need to pass extra arguments to the callback, you can use functools.partial() to wrap the callable.

Parameters:

callback (Callable[[TypeVar(T_Event, bound= Event)], Any]) – a callable that will receive an event object as its only argument.

Return type:

Callable[[TypeVar(T_Event, bound= Event)], Any]

Returns:

the value of callback argument

disconnect(callback)

Disconnects the given callback.

The callback will no longer receive events from this signal.

No action is taken if the callback is not on the list of listener callbacks.

Parameters:

callback (Callable) – the callable to remove

Return type:

None

dispatch(*args, **kwargs)

Create and dispatch an event.

This method constructs an event object and then passes it to dispatch_event() for the actual dispatching.

Parameters:

  • args – positional arguments to the constructor of the associated event class

  • kwargs – keyword arguments to the constructor of the associated event class

Return type:

Awaitable[bool]

Returns:

an awaitable that completes when all the callbacks have been called (and any awaitables waited on) and resolves to True if there were no exceptions raised by the callbacks, False otherwise

dispatch_raw(event)

Dispatch the given event object to all listeners.

Creates a new task in which all listener callbacks are called with the given event as the only argument. Coroutine callbacks are converted to their own respective tasks and waited for concurrently.

Before the dispatching is done, a snapshot of the listeners is taken and the event is only dispatched to those listeners, so adding a listener between the call to this method and the actual dispatching will only affect future calls to this method.

Parameters:

event (Event) – the event object to dispatch

Return type:

Awaitable[bool]

Returns:

an awaitable that completes when all the callbacks have been called (and any awaitables waited on) and resolves to True if there were no exceptions raised by the callbacks, False otherwise

stream_events(filter=None, *, max_queue_size=0)

Shortcut for calling stream_events() with this signal in the first argument.

Return type:

AsyncGenerator[T_Event, None]

wait_event(filter=None)

Shortcut for calling wait_event() with this signal in the first argument.

Return type:

Awaitable[T_Event]

asphalt.core.event.stream_events(signals, filter=None, *, max_queue_size=0)

Return an async generator that yields events from the given signals.

Only events that pass the filter callable (if one has been given) are returned. If no filter function was given, all events are yielded from the generator.

Parameters:

  • signals (Sequence[Signal[T_Event]]) – the signals to get events from

  • filter (Callable[[T_Event], bool] | None) – a callable that takes an event object as an argument and returns True if the event should pass, False if not

  • max_queue_size (int) – maximum size of the queue, after which it will start to drop events

Return type:

AsyncGenerator[T_Event, None]

async asphalt.core.event.wait_event(signals, filter=None)

Wait until any of the given signals dispatches an event that satisfies the filter (if any).

If no filter has been given, the first event dispatched from the signal is returned.

Parameters:

  • signals (Sequence[Signal[T_Event]]) – the signals to get events from

  • filter (Callable[[T_Event], bool] | None) – a callable that takes an event object as an argument and returns True if the event should pass, False if not

Return type:

T_Event

Returns:

the event that was dispatched