Skip to content

event_listener

T = TypeVar('T', bound=BaseEvent) module-attribute

EventListener

Bases: Generic[T]

An event listener that listens for events and handles them.

Attributes:

Name Type Description
on_event Optional[Callable[[T], Optional[BaseEvent | dict]]]

The on_event function that will be called when an event is published. The on_event function should accept an event and return either the event or a dictionary. If the on_event returns None, the event will not be published.
event_types Optional[list[type[T]]]

A list of event types that the event listener should listen for. If not provided, the event listener will listen for all event types.
event_listener_driver Optional[BaseEventListenerDriver]

The driver that will be used to publish events.
Source code in griptape/events/event_listener.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@define
class EventListener(Generic[T]):
    """An event listener that listens for events and handles them.

    Attributes:
        on_event: The on_event function that will be called when an event is published.
            The on_event function should accept an event and return either the event or a dictionary.
            If the on_event returns None, the event will not be published.
        event_types: A list of event types that the event listener should listen for.
            If not provided, the event listener will listen for all event types.
        event_listener_driver: The driver that will be used to publish events.
    """

    on_event: Optional[Callable[[T], Optional[BaseEvent | dict]]] = field(default=None)
    event_types: Optional[list[type[T]]] = field(default=None, kw_only=True)
    event_listener_driver: Optional[BaseEventListenerDriver] = field(default=None, kw_only=True)

    def __enter__(self) -> Self:
        from griptape.events import EventBus

        EventBus.add_event_listener(self)

        return self

    def __exit__(self, type, value, traceback) -> None:  # noqa: ANN001, A002
        from griptape.events import EventBus

        EventBus.remove_event_listener(self)

    def publish_event(self, event: T, *, flush: bool = False) -> None:
        event_types = self.event_types

        if event_types is None or any(isinstance(event, event_type) for event_type in event_types):
            handled_event = event
            if self.on_event is not None:
                handled_event = self.on_event(event)

            if self.event_listener_driver is not None and handled_event is not None:
                self.event_listener_driver.publish_event(handled_event)

        if self.event_listener_driver is not None and flush:
            self.event_listener_driver.flush_events()

event_listener_driver = field(default=None, kw_only=True) class-attribute instance-attribute

event_types = field(default=None, kw_only=True) class-attribute instance-attribute

on_event = field(default=None) class-attribute instance-attribute

__enter__()

Source code in griptape/events/event_listener.py 
35
36
37
38
39
40
def __enter__(self) -> Self:
    from griptape.events import EventBus

    EventBus.add_event_listener(self)

    return self

__exit__(type, value, traceback)

Source code in griptape/events/event_listener.py 
42
43
44
45
def __exit__(self, type, value, traceback) -> None:  # noqa: ANN001, A002
    from griptape.events import EventBus

    EventBus.remove_event_listener(self)

publish_event(event, *, flush=False)

Source code in griptape/events/event_listener.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
def publish_event(self, event: T, *, flush: bool = False) -> None:
    event_types = self.event_types

    if event_types is None or any(isinstance(event, event_type) for event_type in event_types):
        handled_event = event
        if self.on_event is not None:
            handled_event = self.on_event(event)

        if self.event_listener_driver is not None and handled_event is not None:
            self.event_listener_driver.publish_event(handled_event)

    if self.event_listener_driver is not None and flush:
        self.event_listener_driver.flush_events()