Skip to content

API Reference

forgeevent.handlers.base

Base classes for event handlers in the forgeevent framework.

This module provides the abstract base class BaseEventHandler, which defines the interface for handling events. Subclasses must implement the handle method to define how individual events are processed.

BaseEventHandler

Bases: ABC

Abstract base class for event handlers.

This class defines the interface for handling events. Subclasses must implement the handle method to define how individual events are processed.

Methods

dispatch(events: Iterable[Event]) -> None Asynchronously dispatches a collection of events by handling each event. handle(event: Event) -> None Abstract method to handle a single event. Must be implemented by subclasses.

Source code in forgeevent/handlers/base.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
class BaseEventHandler(ABC):
    """
    Abstract base class for event handlers.

    This class defines the interface for handling events.
    Subclasses must implement the `handle` method
    to define how individual events are processed.

    Methods
    -------
    dispatch(events: Iterable[Event]) -> None
        Asynchronously dispatches a collection of events by handling each event.
    handle(event: Event) -> None
        Abstract method to handle a single event. Must be implemented by subclasses.
    """

    async def dispatch(self, events: Iterable[Event[Any]]) -> None:
        """
        Dispatch a collection of events by asynchronously handling each event.

        Parameters
        ----------
        events : Iterable[Event]
            An iterable of Event objects to be handled.
        """
        await asyncio.gather(*[self.handle(event) for event in events])

    @abc.abstractmethod
    async def handle(self, event: Event[Any]) -> None:
        """
        Handle a single event.

        Parameters
        ----------
        event : Event
            The event to handle.
        """
        raise NotImplementedError

dispatch(events) async

Dispatch a collection of events by asynchronously handling each event.

Parameters

events : Iterable[Event] An iterable of Event objects to be handled.

Source code in forgeevent/handlers/base.py 
34
35
36
37
38
39
40
41
42
43
async def dispatch(self, events: Iterable[Event[Any]]) -> None:
    """
    Dispatch a collection of events by asynchronously handling each event.

    Parameters
    ----------
    events : Iterable[Event]
        An iterable of Event objects to be handled.
    """
    await asyncio.gather(*[self.handle(event) for event in events])

handle(event) abstractmethod async

Handle a single event.

Parameters

event : Event The event to handle.

Source code in forgeevent/handlers/base.py 
45
46
47
48
49
50
51
52
53
54
55
@abc.abstractmethod
async def handle(self, event: Event[Any]) -> None:
    """
    Handle a single event.

    Parameters
    ----------
    event : Event
        The event to handle.
    """
    raise NotImplementedError

forgeevent.handlers.local

Local event handler for managing and dispatching events within the application.

This module provides the LocalHandler class, which allows registering event listeners, retrieving registered listeners, and handling events asynchronously or synchronously.

LocalHandler

Bases: BaseEventHandler

Event handler that manages event listeners and dispatches events locally.

This class allows registering listeners for specific event names and handles the execution of those listeners when an event is triggered. It supports both synchronous and asynchronous listeners.

Attributes

_listeners : ClassVar[dict[EventName, list[Callable[..., Any]]]] Dictionary mapping event names to lists of listener functions.

Source code in forgeevent/handlers/local.py 
 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
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class LocalHandler(BaseEventHandler):
    """
    Event handler that manages event listeners and dispatches events locally.

    This class allows registering listeners for specific event names and handles the execution
    of those listeners when an event is triggered. It supports both synchronous and asynchronous
    listeners.

    Attributes
    ----------
    _listeners : ClassVar[dict[EventName, list[Callable[..., Any]]]]
        Dictionary mapping event names to lists of listener functions.
    """

    _listeners: ClassVar[dict[EventName, list[Callable[..., Any]]]] = {}

    def _get_listeners(self, event_name: EventName) -> list[Callable[..., Any]]:
        """
        Retrieve the list of listeners for a specific event name.

        Parameters
        ----------
        event_name : EventName
            The name of the event to get listeners for.

        Returns
        -------
        list[Callable[..., Any]]
            The list of registered listeners for the event name.
        """
        return self._listeners.get(event_name, [])

    def _register_handler(self, event_name: EventName, listener: Callable[..., Any]) -> None:
        """
        Register a listener for a specific event name.

        Parameters
        ----------
        event_name : EventName
            The name of the event to register the listener for.
        listener : Callable[..., Any]
            The function to call when the event is triggered.
        """

        logger.debug(
            'Registering handler for event name: %s with listener: %s', event_name, listener
        )

        logger.debug('Current listeners for %s: %s', event_name, self._listeners)

        self._listeners.setdefault(event_name, []).append(listener)

        logger.debug('Updated listeners for %s: %s', event_name, self._listeners)

    def register(
        self, event_name: EventName, listener: Callable[..., Any] | None = None
    ) -> Callable[..., Any]:
        """
        Register a listener for a specific event name, optionally as a decorator.

        Parameters
        ----------
        event_name : EventName
            The name of the event to listen for.
        listener : Callable[..., Any] | None
            The function to register, or None if used as a decorator.

        Returns
        -------
        Callable[..., Any]
            The registered listener or a decorator.

        Examples
        --------
        As a decorator:
            @handler.register(event_name=MyEvent)
            def on_my_event(event):
                ...

        As a direct call:
            def on_my_event(event):
                ...
            handler.register(event_name=MyEvent, listener=on_my_event)
        """

        def wrapper(_listener: Callable[..., Any]) -> Callable[..., Any]:
            """
            Decorator to register a listener for an event name.
            Args:
                listener (Callable[..., Any]): The function to register as a listener.
            Returns:
                Callable[..., Any]: The wrapped listener function.
            """

            logger.debug('Wrapping listener for event name: %s', event_name)
            self._register_handler(event_name, _listener)
            return _listener

        if listener is None:
            logger.debug('Registering as a decorator when no listener is provided.')
            return wrapper

        logger.debug('Registering listener directly for event name: %s', event_name)
        return wrapper(_listener=listener)

    async def handle(self, event: Event[Any]) -> None:
        """
        Handle a single event.

        Parameters
        ----------
        event : Event
            The event to handle.
        """

        event_name, payload = event
        logger.debug('Handling event: %s with payload: %s', event_name, payload)

        listeners = self._get_listeners(event_name)

        for listener in listeners:
            if asyncio.iscoroutinefunction(listener):
                logger.debug(
                    'Calling async listener: %s for event: %s', listener.__name__, event_name
                )
                await listener(event)
            else:
                logger.debug('Calling sync listener: %s for event: %s', listener, event_name)
                # Run the synchronous listener in a separate thread to avoid blocking the event loop
                loop = asyncio.get_event_loop()
                await loop.run_in_executor(None, listener, event)

handle(event) async

Handle a single event.

Parameters

event : Event The event to handle.

Source code in forgeevent/handlers/local.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
async def handle(self, event: Event[Any]) -> None:
    """
    Handle a single event.

    Parameters
    ----------
    event : Event
        The event to handle.
    """

    event_name, payload = event
    logger.debug('Handling event: %s with payload: %s', event_name, payload)

    listeners = self._get_listeners(event_name)

    for listener in listeners:
        if asyncio.iscoroutinefunction(listener):
            logger.debug(
                'Calling async listener: %s for event: %s', listener.__name__, event_name
            )
            await listener(event)
        else:
            logger.debug('Calling sync listener: %s for event: %s', listener, event_name)
            # Run the synchronous listener in a separate thread to avoid blocking the event loop
            loop = asyncio.get_event_loop()
            await loop.run_in_executor(None, listener, event)

register(event_name, listener=None)

Register a listener for a specific event name, optionally as a decorator.

Parameters

event_name : EventName The name of the event to listen for. listener : Callable[..., Any] | None The function to register, or None if used as a decorator.

Returns

Callable[..., Any] The registered listener or a decorator.

Examples

As a decorator: @handler.register(event_name=MyEvent) def on_my_event(event): ...

As a direct call

def on_my_event(event): ... handler.register(event_name=MyEvent, listener=on_my_event)

Source code in forgeevent/handlers/local.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def register(
    self, event_name: EventName, listener: Callable[..., Any] | None = None
) -> Callable[..., Any]:
    """
    Register a listener for a specific event name, optionally as a decorator.

    Parameters
    ----------
    event_name : EventName
        The name of the event to listen for.
    listener : Callable[..., Any] | None
        The function to register, or None if used as a decorator.

    Returns
    -------
    Callable[..., Any]
        The registered listener or a decorator.

    Examples
    --------
    As a decorator:
        @handler.register(event_name=MyEvent)
        def on_my_event(event):
            ...

    As a direct call:
        def on_my_event(event):
            ...
        handler.register(event_name=MyEvent, listener=on_my_event)
    """

    def wrapper(_listener: Callable[..., Any]) -> Callable[..., Any]:
        """
        Decorator to register a listener for an event name.
        Args:
            listener (Callable[..., Any]): The function to register as a listener.
        Returns:
            Callable[..., Any]: The wrapped listener function.
        """

        logger.debug('Wrapping listener for event name: %s', event_name)
        self._register_handler(event_name, _listener)
        return _listener

    if listener is None:
        logger.debug('Registering as a decorator when no listener is provided.')
        return wrapper

    logger.debug('Registering listener directly for event name: %s', event_name)
    return wrapper(_listener=listener)

forgeevent.typing

Type definitions and aliases for ForgeEvent.

This module provides type hints and utility types for event handling, payloads, and ASGI application interfaces, leveraging Pydantic models and standard Python typing constructs.