fabricatio_core.emitter

Core module that contains the Env class for managing event handling.

Attributes

WILDCARD
Callback	Callback type for event handlers.
EMITTER	The global event emitter instance.

Classes

EventEmitter

An event emitter that supports both exact and wildcard event matching.

Module Contents

fabricatio_core.emitter.WILDCARD = '*'

type fabricatio_core.emitter.Callback = Callable[[T], Coroutine[None, None, None]]

Callback type for event handlers.

class fabricatio_core.emitter.EventEmitter[T](sep: str = '::')

An event emitter that supports both exact and wildcard event matching.

The emitter allows registering event handlers for specific events or patterns containing wildcards (*). It can then emit events and invoke all matching handlers concurrently.

sep = '::'

on(pattern: str, callback: Callback[T]) → Self

Registers an event handler for a specific pattern.

The pattern can be an exact event name or contain wildcards (*) to match multiple events. The callback will be invoked whenever an event matching the pattern is emitted.

Parameters:

• pattern – The event pattern to register the handler for.

• callback – The async callback function to invoke. It must be a coroutine function or return a Future/Task.

Raises:

ValueError – If the pattern is empty.

off(pattern: str) → Self

Removes an event handler for a specific pattern.

The pattern must match the pattern used when registering the handler.

Parameters:

pattern – The event pattern to remove the handler for.

Raises:

ValueError – If the pattern is empty.

async emit(event: str, data: T) → None

Emits an event with the given data to all matching handlers.

This method finds all handlers that match the event pattern (both exact and wildcard matches) and invokes them concurrently with the provided data.

Parameters:

• event – The name of the event to emit.

• data – The data to pass to the event handlers.

Note

The execution of the event handlers is concurrent, and this method will wait for all handlers to complete before returning.

emit_future(event: str, data: T) → asyncio.tasks.Task

Emits an event with the given data to all matching handlers.

This method finds all handlers that match the event pattern (both exact and wildcard matches) and invokes them concurrently with the provided data.

Parameters:

• event – The name of the event to emit.

• data – The data to pass to the event handlers.

Returns:

A future that will be completed when all handlers have been invoked.

fabricatio_core.emitter.EMITTER: EventEmitter[fabricatio_core.models.task.Task]

The global event emitter instance.