Event emitter

Defines the event emitter module used throughout the Gen AI applications.

Authors

Henry Wicaksono (henry.wicaksono@gdplabs.id)

References

NONE

EventEmitter(handlers, event_level=EventLevel.INFO, hooks=None)

Handles events emitting using event handlers with various levels and types.

The EventEmitter class is responsible for handling and emitting events using a list of event handlers. Events are processed based on their severity level and type, with the option to disable specific handlers.

Attributes:

Name	Type	Description
handlers	list[BaseEventHandler]	A list of event handlers to process emitted events.
severity	int	The minimum severity level of events to be processed.

Examples:

Basic usage:

event_emitter = EventEmitter(handlers=[ConsoleEventHandler()])
await event_emitter.emit("Hello, world!")

Emitting an event object:

event_emitter = EventEmitter(handlers=[ConsoleEventHandler()])
event = Event(id="123", value="Hello, world!", level=EventLevel.INFO, type="object", metadata={})
await event_emitter.emit(event)

Using with hooks:

event_emitter = EventEmitter(handlers=[ConsoleEventHandler()], hooks=[JSONStringifyEventHook()])
await event_emitter.emit("Hello, world!")

Using with customized event level:

event_emitter = EventEmitter(handlers=[ConsoleEventHandler()], event_level=EventLevel.DEBUG)
await event_emitter.emit("Hello, world!")

Using with console handler:

event_emitter = EventEmitter.with_console_handler()
await event_emitter.emit("Hello, world!")

Using with print handler:

event_emitter = EventEmitter.with_print_handler()
await event_emitter.emit("Hello, world!")

Using with stream handler:

async def function(event_emitter: EventEmitter):
    await event_emitter.emit("Hello, world!")

event_emitter = EventEmitter.with_stream_handler()
asyncio.create_task(function(event_emitter))
async for event in event_emitter.stream():
    print(event)

Initializes a new instance of the EventEmitter class.

Parameters:

Name	Type	Description	Default
handlers	list[BaseEventHandler]	A list of event handlers to process emitted events.	required
event_level	EventLevel	The minimum severity level of events to be processed. Defaults to EventLevel.INFO.	INFO
hooks	list[BaseEventHook] | None	A list of event hooks to be applied to the emitted events. Defaults to None, in which case the event emitter will not apply any hooks.	None

Raises:

Type	Description
ValueError	If the event handlers list is empty. If an invalid event level is provided.

close() async

Closes all handlers in the handler list.

This method iterates through the list of handlers and calls the close method on each handler.

Returns:

Type	Description
None	None

emit(value, event_level=EventLevel.DEBUG, event_type=EventType.STATUS.value, metadata=None, event_id=None, disabled_handlers=None) async

Emits an event using the configured event handlers.

If the value is an Event object, it is used directly and other parameters are ignored. Otherwise, a new Event object is created with the provided parameters. The event is then passed to all handlers except those listed in disabled_handlers. Events are only processed if their severity level meets or exceeds the EventEmitter's configured level.

Parameters:

Name	Type	Description	Default
value	str | dict[str, Any] | Event	The event value or Event object to emit. If an Event object is provided, other parameters are ignored and the Event is used as-is.	required
event_level	EventLevel	The severity level of the event. Defaults to EventLevel.DEBUG.	DEBUG
event_type	str	The type of event (e.g., "status", "response"). Defaults to EventType.STATUS.value.	value
metadata	dict[str, Any] | None	Additional metadata for the event. Defaults to None, which creates an empty dictionary.	None
event_id	str | None	Unique identifier for the event. Defaults to None, which creates an empty string.	None
disabled_handlers	list[str] | None	Names of handlers to skip for this event. Defaults to None.	None

Raises:

Type	Description
ValueError	If the provided event_level is not a valid EventLevel.

stream()

Streams events from the StreamEventHandler.

This method is a convenience method that calls the stream method on the StreamEventHandler. To use this method, the event emitter must have exactly one StreamEventHandler.

Returns:

Name	Type	Description
AsyncGenerator	AsyncGenerator	An asynchronous generator yielding events from the StreamEventHandler.

with_console_handler(event_level=EventLevel.INFO, hooks=None) classmethod

Creates a new instance of the EventEmitter class with a single ConsoleEventHandler.

Parameters:

Name	Type	Description	Default
event_level	EventLevel	The minimum severity level of events to be processed. Defaults to EventLevel.INFO.	INFO
hooks	list[BaseEventHook] | None	A list of event hooks to be applied to the emitted events. Defaults to None, in which case the event emitter will not apply any hooks.	None

Returns:

Name	Type	Description
EventEmitter	EventEmitter	A new instance of the EventEmitter class with a single ConsoleEventHandler.

with_print_handler(event_level=EventLevel.INFO, hooks=None) classmethod

Creates a new instance of the EventEmitter class with a single PrintEventHandler.

Parameters:

Name	Type	Description	Default
event_level	EventLevel	The minimum severity level of events to be processed. Defaults to EventLevel.INFO.	INFO
hooks	list[BaseEventHook] | None	A list of event hooks to be applied to the emitted events. Defaults to None, in which case the event emitter will not apply any hooks.	None

Returns:

Name	Type	Description
EventEmitter	EventEmitter	A new instance of the EventEmitter class with a single PrintEventHandler.

with_stream_handler(event_level=EventLevel.INFO, hooks=None) classmethod

Creates a new instance of the EventEmitter class with a single StreamEventHandler.

Parameters:

Name	Type	Description	Default
event_level	EventLevel	The minimum severity level of events to be processed. Defaults to EventLevel.INFO.	INFO
hooks	list[BaseEventHook] | None	A list of event hooks to be applied to the emitted events. Defaults to None, in which case the event emitter will not apply any hooks.	None

Returns:

Name	Type	Description
EventEmitter	EventEmitter	A new instance of the EventEmitter class with a single StreamEventHandler.