Skip to content

API Reference

EventEmitter

An EventEmitter class that executes listeners synchronously.

This class allows you to add and remove listeners for specified events and emit those events with arbitrary arguments. When a new listener is added, the EventEmitter emits a "new_listener" event. When an existing listener is removed, the EventEmitter emits a "remove_listener" event.

Events

Event: "new_listener"

The EventEmitter instance will emit its own "new_listener" event before a listener is added to its internal list of listeners. Listeners registered for the "new_listener" event are passed the event name and a reference to the listener being added.

Name Type Descriptions
event Hashable The name of the event being listened for
listener Listenable The event handler function

Event: "remove_listener"

The "remove_listener" event is emitted after the listener is removed.

Name Type Descriptions
event Hashable The name of the event being listened for
listener Listenable The event handler function
Source code in eventemitter/eventemitter.py 
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
class EventEmitter(AbstractEventEmitter[Listenable, Handler]):
    """An `EventEmitter` class that executes listeners synchronously.

    This class allows you to add and remove listeners for specified events and emit those events with arbitrary arguments.
    When a new listener is added, the `EventEmitter` emits a `"new_listener"` event. When an existing listener is removed, the `EventEmitter` emits a `"remove_listener"` event.

    # Events

    **Event: `"new_listener"`**

    The `EventEmitter` instance will emit its own `"new_listener"` event before a `listener` is added to its internal list of listeners.
    Listeners registered for the `"new_listener"` event are passed the `event` name and a reference to the `listener` being added.

    Name       | Type                                  | Descriptions
    ---------- | ------------------------------------- | ----------------------------------------
    `event`    | [Hashable][typing.Hashable]           | The name of the event being listened for
    `listener` | [Listenable][eventemitter.Listenable] | The event handler function


    **Event: `"remove_listener"`**

    The `"remove_listener"` event is emitted after the `listener` is removed.

    Name       | Type                                  | Descriptions
    ---------- | ------------------------------------- | ----------------------------------------
    `event`    | [Hashable][typing.Hashable]           | The name of the event being listened for
    `listener` | [Listenable][eventemitter.Listenable] | The event handler function
    """

    _handler_cls = Handler
    _executor = staticmethod(lambda func, *args, **kwargs: func(*args, **kwargs))

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Initialize an instance of `EventEmitter`.

        Args:
            *args: Arbitrary positional arguments
            **kwargs: Arbitrary keyword arguments
        """
        super().__init__(*args, **kwargs)

    def emit(self, event: Hashable, *args: Any, **kwargs: Any) -> bool:
        """Call each of the listeners registered for the event named `event`, in the order they were registered, passing the supplied arguments to each.

        Args:
            event: The name of the event
            *args: Arbitrary positional arguments
            **kwargs: Arbitrary keyword arguments

        Returns:
            (bool): `True` if the `event` had listeners, `False` otherwise.
        """
        if event not in self._events:
            return False

        for handler in self._events.handlers(event):
            if handler.once:
                self._remove_handler(event, handler)

            handler(*args, **kwargs)

        return True

    def _emit_until_complete(self, event: Hashable, *args: Any, **kwargs: Any) -> None:
        self.emit(event, *args, **kwargs)

__init__ 

__init__(*args: Any, **kwargs: Any) -> None

Initialize an instance of EventEmitter.

Parameters:

Name Type Description Default
*args Any

Arbitrary positional arguments

 ()
**kwargs Any

Arbitrary keyword arguments

 {}
Source code in eventemitter/eventemitter.py 
287
288
289
290
291
292
293
294
def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize an instance of `EventEmitter`.

    Args:
        *args: Arbitrary positional arguments
        **kwargs: Arbitrary keyword arguments
    """
    super().__init__(*args, **kwargs)

emit 

emit(event: Hashable, *args: Any, **kwargs: Any) -> bool

Call each of the listeners registered for the event named event, in the order they were registered, passing the supplied arguments to each.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
*args Any

Arbitrary positional arguments

 ()
**kwargs Any

Arbitrary keyword arguments

 {}

Returns:

Type Description
bool

True if the event had listeners, False otherwise.
Source code in eventemitter/eventemitter.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
def emit(self, event: Hashable, *args: Any, **kwargs: Any) -> bool:
    """Call each of the listeners registered for the event named `event`, in the order they were registered, passing the supplied arguments to each.

    Args:
        event: The name of the event
        *args: Arbitrary positional arguments
        **kwargs: Arbitrary keyword arguments

    Returns:
        (bool): `True` if the `event` had listeners, `False` otherwise.
    """
    if event not in self._events:
        return False

    for handler in self._events.handlers(event):
        if handler.once:
            self._remove_handler(event, handler)

        handler(*args, **kwargs)

    return True

add_listener 

add_listener(event: Hashable, listener: Listenable) -> Self

Add the listener function to the end of the listeners list for the event named event. Multiple calls passing the same combination of event and listener will result in the listener being added, and called, multiple times.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener Listenable

The callback function

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
17
18
19
20
21
22
23
24
25
26
F = TypeVar("F", bound=Union[Listenable, AsyncListenable])  # for functions


# Reference: https://nodejs.org/api/events.html
class AbstractEventEmitter(ABC, EventEmitterProtocol[L], Generic[L, H]):
    """An abstract class for `EventEmitter` classes.

    This class provides a generic implementation for `EventEmitter`s that can manage and emit listeners for events.
    All `EventEmitter`s emit the event `"new_listener"` when new listeners are added and `"remove_listener"` when existing listeners are removed.
    """

prepend_listener 

prepend_listener(
    event: Hashable, listener: Listenable
) -> Self

Add the listener function to the beginning of the listeners list for the event named event. Multiple calls passing the same combination of event and listener will result in the listener being added, and called, multiple times.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener Listenable

The callback function

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
29
30
31
32
33
34
35
36
37
38
39
__slots__ = ("_events",)

_handler_cls: Type[H]

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize an instance of [`AbstractEventEmitter`][eventemitter.AbstractEventEmitter].

    Args:
        *args: Arbitrary positional arguments
        **kwargs: Arbitrary keyword arguments
    """

prepend_once_listener 

prepend_once_listener(
    event: Hashable, listener: Listenable
) -> Self

Add a one-time listener function for the event named event to the beginning of the listeners list. The next time event is triggered, this listener is removed, and then invoked.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener Listenable

The callback function

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
41
42
43
44
45
46
47
48
49
50
    # Reference: https://rhettinger.wordpress.com/2011/05/26/super-considered-super/
    super().__init__(*args, **kwargs)
    self._events: Events[L, H] = Events[L, H]()

def add_listener(self, event: Hashable, listener: L) -> Self:
    """Add the `listener` function to the end of the listeners list for the event named `event`. Multiple calls passing the same combination of `event` and `listener` will result in the `listener` being added, and called, multiple times.

    Args:
        event: The name of the event
        listener: The callback function

events 

events() -> list[Hashable]

Return a list of the events for which the emitter has registered listeners.

Returns:

Type Description
list[Hashable]

A list of the events
Source code in eventemitter/eventemitter.py 
66
67
68
69
70
71
72
    """
    return self._prepend_handler(event, self._handler_cls.from_func(listener, once=False))

def prepend_once_listener(self, event: Hashable, listener: L) -> Self:
    """Add a **one-time** `listener` function for the event named `event` to the *beginning* of the listeners list. The next time `event` is triggered, this `listener` is removed, and then invoked.

    Args:

listeners 

listeners(event: Hashable) -> list[Listenable]

Return a copy of the list of listeners for the event named event.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required

Returns:

Type Description
list[Listenable]

A copy of the list of listeners for the event named event.
Source code in eventemitter/eventemitter.py 
74
75
76
77
78
79
80
81
82
83
        listener: The callback function

    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    return self._prepend_handler(event, self._handler_cls.from_func(listener, once=True))

@abstractmethod
def emit(self, event: Hashable, *args: Any, **kwargs: Any) -> Returns[bool]:
    """Call each of the listeners registered for the event named `event`, passing the supplied arguments to each.

off 

off(event: Hashable, listener: Listenable) -> Self

Alias for remove_listener().

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener Listenable

The listener to remove

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
85
86
87
88
89
90
91
92
93
94
95
    Args:
        event: The name of the event
        *args: Arbitrary positional arguments
        **kwargs: Arbitrary keyword arguments

    Returns:
        (bool): `True` if the `event` had listeners, `False` otherwise.
    """
    raise NotImplementedError()

def events(self) -> List[Hashable]:

on 

on(
    event: Hashable, listener: F | None = None
) -> Self | Callable[[F], F]

Add the listener function to the end of the listeners list for the event named event if a listener is provided, or return a decorator to add the decorated function as a listener if no listener is provided.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener F | None

The callback function

 None

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained if a listener is provided.
Callable[[F], F]

A decorator that adds the decorated function to the listeners list for the event named event.
Source code in eventemitter/eventemitter.py 
102
103
104
105
106
107
108
109
110
111
def listeners(self, event: Hashable) -> List[L]:
    """Return a copy of the list of listeners for the event named `event`.

    Args:
        event: The name of the event

    Returns:
        A copy of the list of listeners for the event named `event`.
    """
    return self._events.listeners(event)

once 

once(
    event: Hashable, listener: F | None = None
) -> Self | Callable[[F], F]

Add a one-time listener function to the end of the listeners list for the event named event if a listener is provided, or return a decorator that adds the decorated function as a one-time listener if no listener is provided.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener F | None

The callback function

 None

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained if a listener is provided.
Callable[[F], F]

A decorator that adds the decorated function as a one-time listener for the event named event.
Source code in eventemitter/eventemitter.py 
120
121
122
123
124
125
126
127
128
129
130
    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    return self.remove_listener(event, listener)

@overload
def on(self, event: Hashable, listener: L) -> Self: ...
@overload
def on(self, event: Hashable) -> Callable[[F], F]: ...

def on(self, event: Hashable, listener: Optional[F] = None) -> Union[Self, Callable[[F], F]]:

remove_all_listeners 

remove_all_listeners(event: Hashable | None = None) -> Self

Remove all listeners, or those of the specified event.

Parameters:

Name Type Description Default
event Hashable | None

The name of the event

 None

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
133
134
135
136
137
138
139
140
Args:
    event: The name of the event
    listener: The callback function

Returns:
    (Self): An instance of the `EventEmitter`, so that calls can be chained if a `listener` is provided.
    (Callable[[F], F]): A decorator that adds the decorated function to the listeners list for the event named `event` if no `listener` is provided.
"""

remove_listener 

remove_listener(
    event: Hashable, listener: Listenable
) -> Self

Remove the specified listener from the listener list for the event named event.

remove_listener() will remove, at most, one instance of a listener from the listener list. If any single listener has been added multiple times to the listener list for the specified event, then remove_listener() must be called multiple times to remove each instance.

Once an event is emitted, all listeners attached to it at the time of emitting are called in order. This implies that any remove_listener() or remove_all_listeners() calls after emitting and before the last listener finishes execution will not remove them from emit() in progress. Subsequent events behave as expected.

When a single function has been added as a handler multiple times for a single event, remove_listener() will remove the most recently added instance.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener Listenable

The listener to remove

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
        self._append_handler(event, self._handler_cls.from_func(listener, once=False))
        return listener

    if listener is not None:
        decorator(listener)

    return decorator if listener is None else self

@overload
def once(self, event: Hashable, listener: L) -> Self: ...
@overload
def once(self, event: Hashable) -> Callable[[F], F]: ...

def once(self, event: Hashable, listener: Optional[F] = None) -> Union[Self, Callable[[F], F]]:
    """Add a **one-time** `listener` function to the end of the listeners list for the event named `event` if a `listener` is provided, or return a decorator that adds the decorated function as a **one-time** `listener` if no `listener` is provided.

    Args:
        event: The name of the event
        listener: The callback function

AsyncIOEventEmitter

An AsyncIOEventEmitter class that executes listeners asynchronously.

This class allows you to add and remove listeners for specified events and emit those events with arbitrary arguments in a non-blocking manner. When a new listener is added, the AsyncIOEventEmitter emits a "new_listener" event. When an existing listener is removed, it emits a "remove_listener" event.

Events

Event: "new_listener"

The AsyncIOEventEmitter instance will emit its own "new_listener" event before a listener is added to its internal list of listeners. Listeners registered for the "new_listener" event are passed the event name and a reference to the listener being added.

Notes
  • A new listener is guaranteed to be added to the internal list of listeners only after all listeners registered for the "new_listener" event have been scheduled and have completed their execution.
  • As a side effect of this implementation, add_listener(), prepend_listener(), prepend_once_listener(), on(), and once() methods operate in a blocking manner.
Name Type Descriptions
event Hashable The name of the event being listened for
listener Listenable | AsyncListenable The event handler function

Event: "remove_listener"

The "remove_listener" event is emitted after the listener is removed.

Name Type Descriptions
event Hashable The name of the event being listened for
listener Listenable | AsyncListenable The event handler function
Notes
  • Similarly to "new_listener" event, remove_listener(), remove_all_listeners(), and off() methods return only after waiting for all "remove_listener" event listeners to complete.
Source code in eventemitter/eventemitter.py 
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
class AsyncIOEventEmitter(AbstractEventEmitter[Union[Listenable, AsyncListenable], AsyncHandler]):
    r"""An `AsyncIOEventEmitter` class that executes listeners asynchronously.

    This class allows you to add and remove listeners for specified events and emit those events with arbitrary arguments in a non-blocking manner.
    When a new listener is added, the `AsyncIOEventEmitter` emits a `"new_listener"` event. When an existing listener is removed, it emits a `"remove_listener"` event.

    # Events

    **Event: `"new_listener"`**

    The `AsyncIOEventEmitter` instance will emit its own `"new_listener"` event before a `listener` is added to its internal list of listeners.
    Listeners registered for the `"new_listener"` event are passed the `event` name and a reference to the `listener` being added.

    Notes:
        - A new listener is **guaranteed** to be added to the internal list of listeners **only after** all listeners registered for the `"new_listener"` event have been scheduled and **have completed their execution**.
        - As a side effect of this implementation, `add_listener()`, `prepend_listener()`, `prepend_once_listener()`, `on()`, and `once()` methods operate in a blocking manner.

    Name       | Type                                                                                         | Descriptions
    ---------- | -------------------------------------------------------------------------------------------- | ----------------------------------------
    `event`    | [`Hashable`][typing.Hashable]                                                                | The name of the event being listened for
    `listener` | [`Listenable`][eventemitter.Listenable] \| [`AsyncListenable`][eventemitter.AsyncListenable] | The event handler function


    **Event: `"remove_listener"`**

    The `"remove_listener"` event is emitted after the `listener` is removed.

    Name       | Type                                                                                         | Descriptions
    ---------- | -------------------------------------------------------------------------------------------- | ----------------------------------------
    `event`    | [`Hashable`][typing.Hashable]                                                                | The name of the event being listened for
    `listener` | [`Listenable`][eventemitter.Listenable] \| [`AsyncListenable`][eventemitter.AsyncListenable] | The event handler function

    Notes:
        - Similarly to `"new_listener"` event, `remove_listener()`, `remove_all_listeners()`, and `off()` methods return **only after** waiting for all `"remove_listener"` event listeners to complete.
    """

    _handler_cls = AsyncHandler

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Initialize an instance of `AsyncIOEventEmitter`.

        Args:
            *args: Arbitrary positional arguments
            **kwargs: Arbitrary keyword arguments
        """
        super().__init__(*args, **kwargs)

    async def emit(self, event: Hashable, *args: Any, **kwargs: Any) -> bool:
        """Call each of the listeners registered for the event named `event`, simultaneously, passing the supplied arguments to each.

        Args:
            event: The name of the event
            *args: Arbitrary positional arguments
            **kwargs: Arbitrary keyword arguments

        Returns:
            (bool): `True` if the `event` had listeners, `False` otherwise.
        """
        if event not in self._events:
            return False

        tasks = set()
        for handler in self._events.handlers(event):
            if handler.once:
                self._remove_handler(event, handler)

            tasks.add(handler(*args, **kwargs))

        await asyncio.gather(*tasks)

        return True

    async def emit_in_order(self, event: Hashable, *args: Any, **kwargs: Any) -> bool:
        """Call each of the listeners registered for the event named `event`, in the order they were registered, passing the supplied arguments to each.

        Args:
            event: The name of the event
            *args: Arbitrary positional arguments
            **kwargs: Arbitrary keyword arguments

        Returns:
            (bool): `True` if the `event` had listeners, `False` otherwise.
        """
        if event not in self._events:
            return False

        for handler in self._events.handlers(event):
            if handler.once:
                self._remove_handler(event, handler)

            await handler(*args, **kwargs)

        return True

    def _emit_until_complete(self, event: Hashable, *args: Any, **kwargs: Any) -> None:
        run_coroutine(self.emit, event, *args, **kwargs)

__init__ 

__init__(*args: Any, **kwargs: Any) -> None

Initialize an instance of AsyncIOEventEmitter.

Parameters:

Name Type Description Default
*args Any

Arbitrary positional arguments

 ()
**kwargs Any

Arbitrary keyword arguments

 {}
Source code in eventemitter/eventemitter.py 
360
361
362
363
364
365
366
367
def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize an instance of `AsyncIOEventEmitter`.

    Args:
        *args: Arbitrary positional arguments
        **kwargs: Arbitrary keyword arguments
    """
    super().__init__(*args, **kwargs)

emit async 

emit(event: Hashable, *args: Any, **kwargs: Any) -> bool

Call each of the listeners registered for the event named event, simultaneously, passing the supplied arguments to each.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
*args Any

Arbitrary positional arguments

 ()
**kwargs Any

Arbitrary keyword arguments

 {}

Returns:

Type Description
bool

True if the event had listeners, False otherwise.
Source code in eventemitter/eventemitter.py 
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
async def emit(self, event: Hashable, *args: Any, **kwargs: Any) -> bool:
    """Call each of the listeners registered for the event named `event`, simultaneously, passing the supplied arguments to each.

    Args:
        event: The name of the event
        *args: Arbitrary positional arguments
        **kwargs: Arbitrary keyword arguments

    Returns:
        (bool): `True` if the `event` had listeners, `False` otherwise.
    """
    if event not in self._events:
        return False

    tasks = set()
    for handler in self._events.handlers(event):
        if handler.once:
            self._remove_handler(event, handler)

        tasks.add(handler(*args, **kwargs))

    await asyncio.gather(*tasks)

    return True

emit_in_order async 

emit_in_order(
    event: Hashable, *args: Any, **kwargs: Any
) -> bool

Call each of the listeners registered for the event named event, in the order they were registered, passing the supplied arguments to each.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
*args Any

Arbitrary positional arguments

 ()
**kwargs Any

Arbitrary keyword arguments

 {}

Returns:

Type Description
bool

True if the event had listeners, False otherwise.
Source code in eventemitter/eventemitter.py 
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
async def emit_in_order(self, event: Hashable, *args: Any, **kwargs: Any) -> bool:
    """Call each of the listeners registered for the event named `event`, in the order they were registered, passing the supplied arguments to each.

    Args:
        event: The name of the event
        *args: Arbitrary positional arguments
        **kwargs: Arbitrary keyword arguments

    Returns:
        (bool): `True` if the `event` had listeners, `False` otherwise.
    """
    if event not in self._events:
        return False

    for handler in self._events.handlers(event):
        if handler.once:
            self._remove_handler(event, handler)

        await handler(*args, **kwargs)

    return True

add_listener 

add_listener(
    event: Hashable, listener: Listenable | AsyncListenable
) -> Self

Add the listener function to the end of the listeners list for the event named event. Multiple calls passing the same combination of event and listener will result in the listener being added, and called, multiple times.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener Listenable | AsyncListenable

The callback function

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
17
18
19
20
21
22
23
24
25
26
F = TypeVar("F", bound=Union[Listenable, AsyncListenable])  # for functions


# Reference: https://nodejs.org/api/events.html
class AbstractEventEmitter(ABC, EventEmitterProtocol[L], Generic[L, H]):
    """An abstract class for `EventEmitter` classes.

    This class provides a generic implementation for `EventEmitter`s that can manage and emit listeners for events.
    All `EventEmitter`s emit the event `"new_listener"` when new listeners are added and `"remove_listener"` when existing listeners are removed.
    """

prepend_listener 

prepend_listener(
    event: Hashable, listener: Listenable | AsyncListenable
) -> Self

Add the listener function to the beginning of the listeners list for the event named event. Multiple calls passing the same combination of event and listener will result in the listener being added, and called, multiple times.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener Listenable | AsyncListenable

The callback function

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
29
30
31
32
33
34
35
36
37
38
39
__slots__ = ("_events",)

_handler_cls: Type[H]

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize an instance of [`AbstractEventEmitter`][eventemitter.AbstractEventEmitter].

    Args:
        *args: Arbitrary positional arguments
        **kwargs: Arbitrary keyword arguments
    """

prepend_once_listener 

prepend_once_listener(
    event: Hashable, listener: Listenable | AsyncListenable
) -> Self

Add a one-time listener function for the event named event to the beginning of the listeners list. The next time event is triggered, this listener is removed, and then invoked.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener Listenable | AsyncListenable

The callback function

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
41
42
43
44
45
46
47
48
49
50
    # Reference: https://rhettinger.wordpress.com/2011/05/26/super-considered-super/
    super().__init__(*args, **kwargs)
    self._events: Events[L, H] = Events[L, H]()

def add_listener(self, event: Hashable, listener: L) -> Self:
    """Add the `listener` function to the end of the listeners list for the event named `event`. Multiple calls passing the same combination of `event` and `listener` will result in the `listener` being added, and called, multiple times.

    Args:
        event: The name of the event
        listener: The callback function

events 

events() -> list[Hashable]

Return a list of the events for which the emitter has registered listeners.

Returns:

Type Description
list[Hashable]

A list of the events
Source code in eventemitter/eventemitter.py 
66
67
68
69
70
71
72
    """
    return self._prepend_handler(event, self._handler_cls.from_func(listener, once=False))

def prepend_once_listener(self, event: Hashable, listener: L) -> Self:
    """Add a **one-time** `listener` function for the event named `event` to the *beginning* of the listeners list. The next time `event` is triggered, this `listener` is removed, and then invoked.

    Args:

listeners 

listeners(
    event: Hashable,
) -> list[Listenable | AsyncListenable]

Return a copy of the list of listeners for the event named event.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required

Returns:

Type Description
list[Listenable | AsyncListenable]

A copy of the list of listeners for the event named event.
Source code in eventemitter/eventemitter.py 
74
75
76
77
78
79
80
81
82
83
        listener: The callback function

    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    return self._prepend_handler(event, self._handler_cls.from_func(listener, once=True))

@abstractmethod
def emit(self, event: Hashable, *args: Any, **kwargs: Any) -> Returns[bool]:
    """Call each of the listeners registered for the event named `event`, passing the supplied arguments to each.

off 

off(
    event: Hashable, listener: Listenable | AsyncListenable
) -> Self

Alias for remove_listener().

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener Listenable | AsyncListenable

The listener to remove

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
85
86
87
88
89
90
91
92
93
94
95
    Args:
        event: The name of the event
        *args: Arbitrary positional arguments
        **kwargs: Arbitrary keyword arguments

    Returns:
        (bool): `True` if the `event` had listeners, `False` otherwise.
    """
    raise NotImplementedError()

def events(self) -> List[Hashable]:

on 

on(
    event: Hashable, listener: F | None = None
) -> Self | Callable[[F], F]

Add the listener function to the end of the listeners list for the event named event if a listener is provided, or return a decorator to add the decorated function as a listener if no listener is provided.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener F | None

The callback function

 None

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained if a listener is provided.
Callable[[F], F]

A decorator that adds the decorated function to the listeners list for the event named event.
Source code in eventemitter/eventemitter.py 
102
103
104
105
106
107
108
109
110
111
def listeners(self, event: Hashable) -> List[L]:
    """Return a copy of the list of listeners for the event named `event`.

    Args:
        event: The name of the event

    Returns:
        A copy of the list of listeners for the event named `event`.
    """
    return self._events.listeners(event)

once 

once(
    event: Hashable, listener: F | None = None
) -> Self | Callable[[F], F]

Add a one-time listener function to the end of the listeners list for the event named event if a listener is provided, or return a decorator that adds the decorated function as a one-time listener if no listener is provided.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener F | None

The callback function

 None

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained if a listener is provided.
Callable[[F], F]

A decorator that adds the decorated function as a one-time listener for the event named event.
Source code in eventemitter/eventemitter.py 
120
121
122
123
124
125
126
127
128
129
130
    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    return self.remove_listener(event, listener)

@overload
def on(self, event: Hashable, listener: L) -> Self: ...
@overload
def on(self, event: Hashable) -> Callable[[F], F]: ...

def on(self, event: Hashable, listener: Optional[F] = None) -> Union[Self, Callable[[F], F]]:

remove_all_listeners 

remove_all_listeners(event: Hashable | None = None) -> Self

Remove all listeners, or those of the specified event.

Parameters:

Name Type Description Default
event Hashable | None

The name of the event

 None

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
133
134
135
136
137
138
139
140
Args:
    event: The name of the event
    listener: The callback function

Returns:
    (Self): An instance of the `EventEmitter`, so that calls can be chained if a `listener` is provided.
    (Callable[[F], F]): A decorator that adds the decorated function to the listeners list for the event named `event` if no `listener` is provided.
"""

remove_listener 

remove_listener(
    event: Hashable, listener: Listenable | AsyncListenable
) -> Self

Remove the specified listener from the listener list for the event named event.

remove_listener() will remove, at most, one instance of a listener from the listener list. If any single listener has been added multiple times to the listener list for the specified event, then remove_listener() must be called multiple times to remove each instance.

Once an event is emitted, all listeners attached to it at the time of emitting are called in order. This implies that any remove_listener() or remove_all_listeners() calls after emitting and before the last listener finishes execution will not remove them from emit() in progress. Subsequent events behave as expected.

When a single function has been added as a handler multiple times for a single event, remove_listener() will remove the most recently added instance.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener Listenable | AsyncListenable

The listener to remove

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
        self._append_handler(event, self._handler_cls.from_func(listener, once=False))
        return listener

    if listener is not None:
        decorator(listener)

    return decorator if listener is None else self

@overload
def once(self, event: Hashable, listener: L) -> Self: ...
@overload
def once(self, event: Hashable) -> Callable[[F], F]: ...

def once(self, event: Hashable, listener: Optional[F] = None) -> Union[Self, Callable[[F], F]]:
    """Add a **one-time** `listener` function to the end of the listeners list for the event named `event` if a `listener` is provided, or return a decorator that adds the decorated function as a **one-time** `listener` if no `listener` is provided.

    Args:
        event: The name of the event
        listener: The callback function

EventEmitterProtocol

A protocol that describes the structural requirements for an EventEmitter class.

Source code in eventemitter/protocol.py 
 14
 15
 16
 17
 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
 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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class EventEmitterProtocol(Protocol[L]):
    """A protocol that describes the structural requirements for an `EventEmitter` class."""

    def add_listener(self, event: Hashable, listener: L) -> Self:
        """Add the `listener` function to the end of the listeners list for the event named `event`. Multiple calls passing the same combination of `event` and `listener` will result in the `listener` being added, and called, multiple times.

        Args:
            event: The name of the event
            listener: The callback function

        Returns:
            An instance of the `EventEmitter`, so that calls can be chained.
        """
        ...

    def prepend_listener(self, event: Hashable, listener: L) -> Self:
        """Add the `listener` function to the *beginning* of the listeners list for the event named `event`. Multiple calls passing the same combination of `event` and `listener` will result in the `listener` being added, and called, multiple times.

        Args:
            event: The name of the event
            listener: The callback function

        Returns:
            An instance of the `EventEmitter`, so that calls can be chained.
        """
        ...

    def prepend_once_listener(self, event: Hashable, listener: L) -> Self:
        """Add a **one-time** `listener` function for the event named `event` to the *beginning* of the listeners list. The next time `event` is triggered, this `listener` is removed, and then invoked.

        Args:
            event: The name of the event
            listener: The callback function

        Returns:
            An instance of the `EventEmitter`, so that calls can be chained.
        """
        ...

    def emit(self, event: Hashable, *args: Any, **kwargs: Any) -> Returns[bool]:
        """Call each of the listeners registered for the event named `event`, passing the supplied arguments to each.

        Args:
            event: The name of the event
            *args: Arbitrary positional arguments
            **kwargs: Arbitrary keyword arguments

        Returns:
            (bool): `True` if the `event` had listeners, `False` otherwise.
        """
        ...

    def events(self) -> List[Hashable]:
        """Return a list of the events for which the emitter has registered listeners.

        Returns:
            A list of the `event`s
        """
        ...

    def listeners(self, event: Hashable) -> List[L]:
        """Return a copy of the list of listeners for the event named `event`.

        Args:
            event: The name of the event

        Returns:
           A copy of the list of listeners for the event named `event`.
        """
        ...

    def off(self, event: Hashable, listener: L) -> Self:
        """Alias for `remove_listener()`.

        Args:
            event: The name of the event
            listener: The listener to remove

        Returns:
            An instance of the `EventEmitter`, so that calls can be chained.
        """
        ...

    @overload
    def on(self, event: Hashable, listener: L) -> Self: ...
    @overload
    def on(self, event: Hashable) -> Callable[[F], F]: ...

    def on(self, event: Hashable, listener: Optional[F] = None) -> Union[Self, Callable[[F], F]]:
        """Add the `listener` function to the end of the listeners list for the event named `event` if a `listener` is provided, or return a decorator to add the decorated function as a `listener` if no `listener` is provided.

        Args:
            event: The name of the event
            listener: The callback function

        Returns:
            (Self): An instance of the `EventEmitter`, so that calls can be chained if a `listener` is provided.
            (Callable[[F], F]): A decorator that adds the decorated function to the listeners list for the event named `event`.
        """
        ...

    @overload
    def once(self, event: Hashable, listener: L) -> Self: ...
    @overload
    def once(self, event: Hashable) -> Callable[[F], F]: ...

    def once(self, event: Hashable, listener: Optional[F] = None) -> Union[Self, Callable[[F], F]]:
        """Add a **one-time** `listener` function to the end of the listeners list for the event named `event` if a `listener` is provided, or return a decorator that adds the decorated function as a **one-time** `listener` if no `listener` is provided.

        Args:
            event: The name of the event
            listener: The callback function

        Returns:
            (Self): An instance of the `EventEmitter`, so that calls can be chained if a `listener` is provided.
            (Callable[[F], F]): A decorator that adds the decorated function as a **one-time** `listener` for the event named `event`.
        """
        ...

    def remove_all_listeners(self, event: Optional[Hashable] = None) -> Self:
        """Remove all listeners, or those of the specified `event`.

        Args:
            event: The name of the event

        Returns:
            An instance of the `EventEmitter`, so that calls can be chained.
        """
        ...

    def remove_listener(self, event: Hashable, listener: L) -> Self:
        """Remove the specified `listener` from the listener list for the event named `event`.

           `remove_listener()` will remove, at most, one instance of a listener from the listener list.
           If any single listener has been added multiple times to the listener list for the specified `event`, then `remove_listener()` must be called multiple times to remove each instance.

           Once an event is emitted, all listeners attached to it at the time of emitting are called in order.
           This implies that any `remove_listener()` or `remove_all_listeners()` calls after emitting and before the last listener finishes execution will not remove them from emit() in progress.
           Subsequent events behave as expected.

           When a single function has been added as a handler multiple times for a single event, `remove_listener()` will remove the most recently added instance.

        Args:
            event: The name of the event
            listener: The listener to remove

        Returns:
            An instance of the `EventEmitter`, so that calls can be chained.
        """
        ...

add_listener 

add_listener(event: Hashable, listener: L) -> Self

Add the listener function to the end of the listeners list for the event named event. Multiple calls passing the same combination of event and listener will result in the listener being added, and called, multiple times.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener L

The callback function

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/protocol.py 
17
18
19
20
21
22
23
24
25
26
27
def add_listener(self, event: Hashable, listener: L) -> Self:
    """Add the `listener` function to the end of the listeners list for the event named `event`. Multiple calls passing the same combination of `event` and `listener` will result in the `listener` being added, and called, multiple times.

    Args:
        event: The name of the event
        listener: The callback function

    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    ...

prepend_listener 

prepend_listener(event: Hashable, listener: L) -> Self

Add the listener function to the beginning of the listeners list for the event named event. Multiple calls passing the same combination of event and listener will result in the listener being added, and called, multiple times.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener L

The callback function

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/protocol.py 
29
30
31
32
33
34
35
36
37
38
39
def prepend_listener(self, event: Hashable, listener: L) -> Self:
    """Add the `listener` function to the *beginning* of the listeners list for the event named `event`. Multiple calls passing the same combination of `event` and `listener` will result in the `listener` being added, and called, multiple times.

    Args:
        event: The name of the event
        listener: The callback function

    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    ...

prepend_once_listener 

prepend_once_listener(
    event: Hashable, listener: L
) -> Self

Add a one-time listener function for the event named event to the beginning of the listeners list. The next time event is triggered, this listener is removed, and then invoked.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener L

The callback function

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/protocol.py 
41
42
43
44
45
46
47
48
49
50
51
def prepend_once_listener(self, event: Hashable, listener: L) -> Self:
    """Add a **one-time** `listener` function for the event named `event` to the *beginning* of the listeners list. The next time `event` is triggered, this `listener` is removed, and then invoked.

    Args:
        event: The name of the event
        listener: The callback function

    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    ...

emit 

emit(
    event: Hashable, *args: Any, **kwargs: Any
) -> Returns[bool]

Call each of the listeners registered for the event named event, passing the supplied arguments to each.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
*args Any

Arbitrary positional arguments

 ()
**kwargs Any

Arbitrary keyword arguments

 {}

Returns:

Type Description
bool

True if the event had listeners, False otherwise.
Source code in eventemitter/protocol.py 
53
54
55
56
57
58
59
60
61
62
63
64
def emit(self, event: Hashable, *args: Any, **kwargs: Any) -> Returns[bool]:
    """Call each of the listeners registered for the event named `event`, passing the supplied arguments to each.

    Args:
        event: The name of the event
        *args: Arbitrary positional arguments
        **kwargs: Arbitrary keyword arguments

    Returns:
        (bool): `True` if the `event` had listeners, `False` otherwise.
    """
    ...

events 

events() -> list[Hashable]

Return a list of the events for which the emitter has registered listeners.

Returns:

Type Description
list[Hashable]

A list of the events
Source code in eventemitter/protocol.py 
66
67
68
69
70
71
72
def events(self) -> List[Hashable]:
    """Return a list of the events for which the emitter has registered listeners.

    Returns:
        A list of the `event`s
    """
    ...

listeners 

listeners(event: Hashable) -> list[L]

Return a copy of the list of listeners for the event named event.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required

Returns:

Type Description
list[L]

A copy of the list of listeners for the event named event.
Source code in eventemitter/protocol.py 
74
75
76
77
78
79
80
81
82
83
def listeners(self, event: Hashable) -> List[L]:
    """Return a copy of the list of listeners for the event named `event`.

    Args:
        event: The name of the event

    Returns:
       A copy of the list of listeners for the event named `event`.
    """
    ...

off 

off(event: Hashable, listener: L) -> Self

Alias for remove_listener().

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener L

The listener to remove

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/protocol.py 
85
86
87
88
89
90
91
92
93
94
95
def off(self, event: Hashable, listener: L) -> Self:
    """Alias for `remove_listener()`.

    Args:
        event: The name of the event
        listener: The listener to remove

    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    ...

on 

on(
    event: Hashable, listener: F | None = None
) -> Self | Callable[[F], F]

Add the listener function to the end of the listeners list for the event named event if a listener is provided, or return a decorator to add the decorated function as a listener if no listener is provided.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener F | None

The callback function

 None

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained if a listener is provided.
Callable[[F], F]

A decorator that adds the decorated function to the listeners list for the event named event.
Source code in eventemitter/protocol.py 
102
103
104
105
106
107
108
109
110
111
112
113
def on(self, event: Hashable, listener: Optional[F] = None) -> Union[Self, Callable[[F], F]]:
    """Add the `listener` function to the end of the listeners list for the event named `event` if a `listener` is provided, or return a decorator to add the decorated function as a `listener` if no `listener` is provided.

    Args:
        event: The name of the event
        listener: The callback function

    Returns:
        (Self): An instance of the `EventEmitter`, so that calls can be chained if a `listener` is provided.
        (Callable[[F], F]): A decorator that adds the decorated function to the listeners list for the event named `event`.
    """
    ...

once 

once(
    event: Hashable, listener: F | None = None
) -> Self | Callable[[F], F]

Add a one-time listener function to the end of the listeners list for the event named event if a listener is provided, or return a decorator that adds the decorated function as a one-time listener if no listener is provided.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener F | None

The callback function

 None

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained if a listener is provided.
Callable[[F], F]

A decorator that adds the decorated function as a one-time listener for the event named event.
Source code in eventemitter/protocol.py 
120
121
122
123
124
125
126
127
128
129
130
131
def once(self, event: Hashable, listener: Optional[F] = None) -> Union[Self, Callable[[F], F]]:
    """Add a **one-time** `listener` function to the end of the listeners list for the event named `event` if a `listener` is provided, or return a decorator that adds the decorated function as a **one-time** `listener` if no `listener` is provided.

    Args:
        event: The name of the event
        listener: The callback function

    Returns:
        (Self): An instance of the `EventEmitter`, so that calls can be chained if a `listener` is provided.
        (Callable[[F], F]): A decorator that adds the decorated function as a **one-time** `listener` for the event named `event`.
    """
    ...

remove_all_listeners 

remove_all_listeners(event: Hashable | None = None) -> Self

Remove all listeners, or those of the specified event.

Parameters:

Name Type Description Default
event Hashable | None

The name of the event

 None

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/protocol.py 
133
134
135
136
137
138
139
140
141
142
def remove_all_listeners(self, event: Optional[Hashable] = None) -> Self:
    """Remove all listeners, or those of the specified `event`.

    Args:
        event: The name of the event

    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    ...

remove_listener 

remove_listener(event: Hashable, listener: L) -> Self

Remove the specified listener from the listener list for the event named event.

remove_listener() will remove, at most, one instance of a listener from the listener list. If any single listener has been added multiple times to the listener list for the specified event, then remove_listener() must be called multiple times to remove each instance.

Once an event is emitted, all listeners attached to it at the time of emitting are called in order. This implies that any remove_listener() or remove_all_listeners() calls after emitting and before the last listener finishes execution will not remove them from emit() in progress. Subsequent events behave as expected.

When a single function has been added as a handler multiple times for a single event, remove_listener() will remove the most recently added instance.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener L

The listener to remove

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/protocol.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def remove_listener(self, event: Hashable, listener: L) -> Self:
    """Remove the specified `listener` from the listener list for the event named `event`.

       `remove_listener()` will remove, at most, one instance of a listener from the listener list.
       If any single listener has been added multiple times to the listener list for the specified `event`, then `remove_listener()` must be called multiple times to remove each instance.

       Once an event is emitted, all listeners attached to it at the time of emitting are called in order.
       This implies that any `remove_listener()` or `remove_all_listeners()` calls after emitting and before the last listener finishes execution will not remove them from emit() in progress.
       Subsequent events behave as expected.

       When a single function has been added as a handler multiple times for a single event, `remove_listener()` will remove the most recently added instance.

    Args:
        event: The name of the event
        listener: The listener to remove

    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    ...

AbstractEventEmitter

An abstract class for EventEmitter classes.

This class provides a generic implementation for EventEmitters that can manage and emit listeners for events. All EventEmitters emit the event "new_listener" when new listeners are added and "remove_listener" when existing listeners are removed.

Source code in eventemitter/eventemitter.py 
 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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
class AbstractEventEmitter(ABC, EventEmitterProtocol[L], Generic[L, H]):
    """An abstract class for `EventEmitter` classes.

    This class provides a generic implementation for `EventEmitter`s that can manage and emit listeners for events.
    All `EventEmitter`s emit the event `"new_listener"` when new listeners are added and `"remove_listener"` when existing listeners are removed.
    """

    __slots__ = ("_events",)

    _handler_cls: Type[H]

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Initialize an instance of [`AbstractEventEmitter`][eventemitter.AbstractEventEmitter].

        Args:
            *args: Arbitrary positional arguments
            **kwargs: Arbitrary keyword arguments
        """
        # To support cooperative multiple inheritance
        # Reference: https://rhettinger.wordpress.com/2011/05/26/super-considered-super/
        super().__init__(*args, **kwargs)
        self._events: Events[L, H] = Events[L, H]()

    def add_listener(self, event: Hashable, listener: L) -> Self:
        """Add the `listener` function to the end of the listeners list for the event named `event`. Multiple calls passing the same combination of `event` and `listener` will result in the `listener` being added, and called, multiple times.

        Args:
            event: The name of the event
            listener: The callback function

        Returns:
            An instance of the `EventEmitter`, so that calls can be chained.
        """
        return self._append_handler(event, self._handler_cls.from_func(listener, once=False))

    def prepend_listener(self, event: Hashable, listener: L) -> Self:
        """Add the `listener` function to the *beginning* of the listeners list for the event named `event`. Multiple calls passing the same combination of `event` and `listener` will result in the `listener` being added, and called, multiple times.

        Args:
            event: The name of the event
            listener: The callback function

        Returns:
            An instance of the `EventEmitter`, so that calls can be chained.
        """
        return self._prepend_handler(event, self._handler_cls.from_func(listener, once=False))

    def prepend_once_listener(self, event: Hashable, listener: L) -> Self:
        """Add a **one-time** `listener` function for the event named `event` to the *beginning* of the listeners list. The next time `event` is triggered, this `listener` is removed, and then invoked.

        Args:
            event: The name of the event
            listener: The callback function

        Returns:
            An instance of the `EventEmitter`, so that calls can be chained.
        """
        return self._prepend_handler(event, self._handler_cls.from_func(listener, once=True))

    @abstractmethod
    def emit(self, event: Hashable, *args: Any, **kwargs: Any) -> Returns[bool]:
        """Call each of the listeners registered for the event named `event`, passing the supplied arguments to each.

        Args:
            event: The name of the event
            *args: Arbitrary positional arguments
            **kwargs: Arbitrary keyword arguments

        Returns:
            (bool): `True` if the `event` had listeners, `False` otherwise.
        """
        raise NotImplementedError()

    def events(self) -> List[Hashable]:
        """Return a list of the events for which the emitter has registered listeners.

        Returns:
            A list of the `event`s
        """
        return list(self._events.keys())

    def listeners(self, event: Hashable) -> List[L]:
        """Return a copy of the list of listeners for the event named `event`.

        Args:
            event: The name of the event

        Returns:
            A copy of the list of listeners for the event named `event`.
        """
        return self._events.listeners(event)

    def off(self, event: Hashable, listener: L) -> Self:
        """Alias for `remove_listener()`.

        Args:
            event: The name of the event
            listener: The listener to remove

        Returns:
            An instance of the `EventEmitter`, so that calls can be chained.
        """
        return self.remove_listener(event, listener)

    @overload
    def on(self, event: Hashable, listener: L) -> Self: ...
    @overload
    def on(self, event: Hashable) -> Callable[[F], F]: ...

    def on(self, event: Hashable, listener: Optional[F] = None) -> Union[Self, Callable[[F], F]]:
        """Add the `listener` function to the end of the listeners list for the event named `event` if a `listener` is provided, or return a decorator to add the decorated function as a `listener` if no `listener` is provided.

        Args:
            event: The name of the event
            listener: The callback function

        Returns:
            (Self): An instance of the `EventEmitter`, so that calls can be chained if a `listener` is provided.
            (Callable[[F], F]): A decorator that adds the decorated function to the listeners list for the event named `event` if no `listener` is provided.
        """

        def decorator(listener: F) -> F:
            self._append_handler(event, self._handler_cls.from_func(listener, once=False))
            return listener

        if listener is not None:
            decorator(listener)

        return decorator if listener is None else self

    @overload
    def once(self, event: Hashable, listener: L) -> Self: ...
    @overload
    def once(self, event: Hashable) -> Callable[[F], F]: ...

    def once(self, event: Hashable, listener: Optional[F] = None) -> Union[Self, Callable[[F], F]]:
        """Add a **one-time** `listener` function to the end of the listeners list for the event named `event` if a `listener` is provided, or return a decorator that adds the decorated function as a **one-time** `listener` if no `listener` is provided.

        Args:
            event: The name of the event
            listener: The callback function

        Returns:
            (Self): An instance of the `EventEmitter`, so that calls can be chained if a `listener` is provided.
            (Callable[[F], F]): A decorator that adds the decorated function as a **one-time** `listener` for the event named `event` if no `listener` is provided.
        """

        def decorator(listener: F) -> F:
            self._append_handler(event, self._handler_cls.from_func(listener, once=True))
            return listener

        if listener is not None:
            decorator(listener)

        return decorator if listener is None else self

    def remove_all_listeners(self, event: Optional[Hashable] = None) -> Self:
        """Remove all listeners, or those of the specified `event`.

        Args:
            event: The name of the event

        Returns:
            An instance of the `EventEmitter`, so that calls can be chained.
        """
        if event is None:
            for event in self.events():
                self.remove_all_listeners(event)
        else:
            for handler in reversed(self._events.handlers(event)):
                self._remove_handler(event, handler)

        return self

    def remove_listener(self, event: Hashable, listener: L) -> Self:
        """Remove the specified `listener` from the listener list for the event named `event`.

           `remove_listener()` will remove, at most, one instance of a listener from the listener list.
           If any single listener has been added multiple times to the listener list for the specified `event`, then `remove_listener()` must be called multiple times to remove each instance.

           Once an event is emitted, all listeners attached to it at the time of emitting are called in order.
           This implies that any `remove_listener()` or `remove_all_listeners()` calls after emitting and before the last listener finishes execution will not remove them from emit() in progress.
           Subsequent events behave as expected.

           When a single function has been added as a handler multiple times for a single event, `remove_listener()` will remove the most recently added instance.

        Args:
            event: The name of the event
            listener: The listener to remove

        Returns:
            An instance of the `EventEmitter`, so that calls can be chained.
        """
        return self._remove_handler(event, listener)

    def _append_handler(self, event: Hashable, handler: H) -> Self:
        self._emit_until_complete("new_listener", event, handler.func)
        self._events[event].append(handler)
        return self

    def _prepend_handler(self, event: Hashable, handler: H) -> Self:
        self._emit_until_complete("new_listener", event, handler.func)
        self._events[event].prepend(handler)
        return self

    @overload
    def _remove_handler(self, event: Hashable, target: H) -> Self: ...
    @overload
    def _remove_handler(self, event: Hashable, target: Union[Listenable, AsyncListenable]) -> Self: ...

    def _remove_handler(self, event: Hashable, target: Union[H, Listenable, AsyncListenable]) -> Self:
        if event not in self._events:
            return self

        try:
            if isinstance(target, self._handler_cls):
                handler = self._events[event].remove(target, last=True)
            else:
                handler = self._events[event].remove_by_id(target, last=True)

            self._emit_until_complete("remove_listener", event, handler.func)
        except ValueError:
            pass

        if not self._events[event]:
            del self._events[event]

        return self

    @abstractmethod
    def _emit_until_complete(self, event: Hashable, *args: Any, **kwargs: Any) -> None: ...

__init__ 

__init__(*args: Any, **kwargs: Any) -> None

Initialize an instance of AbstractEventEmitter.

Parameters:

Name Type Description Default
*args Any

Arbitrary positional arguments

 ()
**kwargs Any

Arbitrary keyword arguments

 {}
Source code in eventemitter/eventemitter.py 
33
34
35
36
37
38
39
40
41
42
43
def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize an instance of [`AbstractEventEmitter`][eventemitter.AbstractEventEmitter].

    Args:
        *args: Arbitrary positional arguments
        **kwargs: Arbitrary keyword arguments
    """
    # To support cooperative multiple inheritance
    # Reference: https://rhettinger.wordpress.com/2011/05/26/super-considered-super/
    super().__init__(*args, **kwargs)
    self._events: Events[L, H] = Events[L, H]()

add_listener 

add_listener(event: Hashable, listener: L) -> Self

Add the listener function to the end of the listeners list for the event named event. Multiple calls passing the same combination of event and listener will result in the listener being added, and called, multiple times.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener L

The callback function

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
45
46
47
48
49
50
51
52
53
54
55
def add_listener(self, event: Hashable, listener: L) -> Self:
    """Add the `listener` function to the end of the listeners list for the event named `event`. Multiple calls passing the same combination of `event` and `listener` will result in the `listener` being added, and called, multiple times.

    Args:
        event: The name of the event
        listener: The callback function

    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    return self._append_handler(event, self._handler_cls.from_func(listener, once=False))

prepend_listener 

prepend_listener(event: Hashable, listener: L) -> Self

Add the listener function to the beginning of the listeners list for the event named event. Multiple calls passing the same combination of event and listener will result in the listener being added, and called, multiple times.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener L

The callback function

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
57
58
59
60
61
62
63
64
65
66
67
def prepend_listener(self, event: Hashable, listener: L) -> Self:
    """Add the `listener` function to the *beginning* of the listeners list for the event named `event`. Multiple calls passing the same combination of `event` and `listener` will result in the `listener` being added, and called, multiple times.

    Args:
        event: The name of the event
        listener: The callback function

    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    return self._prepend_handler(event, self._handler_cls.from_func(listener, once=False))

prepend_once_listener 

prepend_once_listener(
    event: Hashable, listener: L
) -> Self

Add a one-time listener function for the event named event to the beginning of the listeners list. The next time event is triggered, this listener is removed, and then invoked.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener L

The callback function

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
69
70
71
72
73
74
75
76
77
78
79
def prepend_once_listener(self, event: Hashable, listener: L) -> Self:
    """Add a **one-time** `listener` function for the event named `event` to the *beginning* of the listeners list. The next time `event` is triggered, this `listener` is removed, and then invoked.

    Args:
        event: The name of the event
        listener: The callback function

    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    return self._prepend_handler(event, self._handler_cls.from_func(listener, once=True))

emit abstractmethod 

emit(
    event: Hashable, *args: Any, **kwargs: Any
) -> Returns[bool]

Call each of the listeners registered for the event named event, passing the supplied arguments to each.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
*args Any

Arbitrary positional arguments

 ()
**kwargs Any

Arbitrary keyword arguments

 {}

Returns:

Type Description
bool

True if the event had listeners, False otherwise.
Source code in eventemitter/eventemitter.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
@abstractmethod
def emit(self, event: Hashable, *args: Any, **kwargs: Any) -> Returns[bool]:
    """Call each of the listeners registered for the event named `event`, passing the supplied arguments to each.

    Args:
        event: The name of the event
        *args: Arbitrary positional arguments
        **kwargs: Arbitrary keyword arguments

    Returns:
        (bool): `True` if the `event` had listeners, `False` otherwise.
    """
    raise NotImplementedError()

events 

events() -> list[Hashable]

Return a list of the events for which the emitter has registered listeners.

Returns:

Type Description
list[Hashable]

A list of the events
Source code in eventemitter/eventemitter.py 
 95
 96
 97
 98
 99
100
101
def events(self) -> List[Hashable]:
    """Return a list of the events for which the emitter has registered listeners.

    Returns:
        A list of the `event`s
    """
    return list(self._events.keys())

listeners 

listeners(event: Hashable) -> list[L]

Return a copy of the list of listeners for the event named event.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required

Returns:

Type Description
list[L]

A copy of the list of listeners for the event named event.
Source code in eventemitter/eventemitter.py 
103
104
105
106
107
108
109
110
111
112
def listeners(self, event: Hashable) -> List[L]:
    """Return a copy of the list of listeners for the event named `event`.

    Args:
        event: The name of the event

    Returns:
        A copy of the list of listeners for the event named `event`.
    """
    return self._events.listeners(event)

off 

off(event: Hashable, listener: L) -> Self

Alias for remove_listener().

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener L

The listener to remove

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
114
115
116
117
118
119
120
121
122
123
124
def off(self, event: Hashable, listener: L) -> Self:
    """Alias for `remove_listener()`.

    Args:
        event: The name of the event
        listener: The listener to remove

    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    return self.remove_listener(event, listener)

on 

on(
    event: Hashable, listener: F | None = None
) -> Self | Callable[[F], F]

Add the listener function to the end of the listeners list for the event named event if a listener is provided, or return a decorator to add the decorated function as a listener if no listener is provided.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener F | None

The callback function

 None

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained if a listener is provided.
Callable[[F], F]

A decorator that adds the decorated function to the listeners list for the event named event if no listener is provided.
Source code in eventemitter/eventemitter.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def on(self, event: Hashable, listener: Optional[F] = None) -> Union[Self, Callable[[F], F]]:
    """Add the `listener` function to the end of the listeners list for the event named `event` if a `listener` is provided, or return a decorator to add the decorated function as a `listener` if no `listener` is provided.

    Args:
        event: The name of the event
        listener: The callback function

    Returns:
        (Self): An instance of the `EventEmitter`, so that calls can be chained if a `listener` is provided.
        (Callable[[F], F]): A decorator that adds the decorated function to the listeners list for the event named `event` if no `listener` is provided.
    """

    def decorator(listener: F) -> F:
        self._append_handler(event, self._handler_cls.from_func(listener, once=False))
        return listener

    if listener is not None:
        decorator(listener)

    return decorator if listener is None else self

once 

once(
    event: Hashable, listener: F | None = None
) -> Self | Callable[[F], F]

Add a one-time listener function to the end of the listeners list for the event named event if a listener is provided, or return a decorator that adds the decorated function as a one-time listener if no listener is provided.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener F | None

The callback function

 None

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained if a listener is provided.
Callable[[F], F]

A decorator that adds the decorated function as a one-time listener for the event named event if no listener is provided.
Source code in eventemitter/eventemitter.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def once(self, event: Hashable, listener: Optional[F] = None) -> Union[Self, Callable[[F], F]]:
    """Add a **one-time** `listener` function to the end of the listeners list for the event named `event` if a `listener` is provided, or return a decorator that adds the decorated function as a **one-time** `listener` if no `listener` is provided.

    Args:
        event: The name of the event
        listener: The callback function

    Returns:
        (Self): An instance of the `EventEmitter`, so that calls can be chained if a `listener` is provided.
        (Callable[[F], F]): A decorator that adds the decorated function as a **one-time** `listener` for the event named `event` if no `listener` is provided.
    """

    def decorator(listener: F) -> F:
        self._append_handler(event, self._handler_cls.from_func(listener, once=True))
        return listener

    if listener is not None:
        decorator(listener)

    return decorator if listener is None else self

remove_all_listeners 

remove_all_listeners(event: Hashable | None = None) -> Self

Remove all listeners, or those of the specified event.

Parameters:

Name Type Description Default
event Hashable | None

The name of the event

 None

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def remove_all_listeners(self, event: Optional[Hashable] = None) -> Self:
    """Remove all listeners, or those of the specified `event`.

    Args:
        event: The name of the event

    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    if event is None:
        for event in self.events():
            self.remove_all_listeners(event)
    else:
        for handler in reversed(self._events.handlers(event)):
            self._remove_handler(event, handler)

    return self

remove_listener 

remove_listener(event: Hashable, listener: L) -> Self

Remove the specified listener from the listener list for the event named event.

remove_listener() will remove, at most, one instance of a listener from the listener list. If any single listener has been added multiple times to the listener list for the specified event, then remove_listener() must be called multiple times to remove each instance.

Once an event is emitted, all listeners attached to it at the time of emitting are called in order. This implies that any remove_listener() or remove_all_listeners() calls after emitting and before the last listener finishes execution will not remove them from emit() in progress. Subsequent events behave as expected.

When a single function has been added as a handler multiple times for a single event, remove_listener() will remove the most recently added instance.

Parameters:

Name Type Description Default
event Hashable

The name of the event

 required
listener L

The listener to remove

 required

Returns:

Type Description
Self

An instance of the EventEmitter, so that calls can be chained.
Source code in eventemitter/eventemitter.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def remove_listener(self, event: Hashable, listener: L) -> Self:
    """Remove the specified `listener` from the listener list for the event named `event`.

       `remove_listener()` will remove, at most, one instance of a listener from the listener list.
       If any single listener has been added multiple times to the listener list for the specified `event`, then `remove_listener()` must be called multiple times to remove each instance.

       Once an event is emitted, all listeners attached to it at the time of emitting are called in order.
       This implies that any `remove_listener()` or `remove_all_listeners()` calls after emitting and before the last listener finishes execution will not remove them from emit() in progress.
       Subsequent events behave as expected.

       When a single function has been added as a handler multiple times for a single event, `remove_listener()` will remove the most recently added instance.

    Args:
        event: The name of the event
        listener: The listener to remove

    Returns:
        An instance of the `EventEmitter`, so that calls can be chained.
    """
    return self._remove_handler(event, listener)

AsyncCallable module-attribute 

AsyncCallable = Callable[P, Coroutine[Any, Any, R]]

Listenable module-attribute 

Listenable = Callable[..., None]

AsyncListenable module-attribute 

AsyncListenable = AsyncCallable[..., None]