This module is a simple event manager that works with decorators. I would like feedback on:

  • Bugs
  • Any ideas to get rid of the classevent decorator, which complicates the usage, and is error-prone.
  • The readability of the code
  • The quality and thoroughness of the unit tests


"""A module that defines decorators to transform functions and methods
into events"""

from types import MethodType
from typing import Callable, TypeVar, Generic, Any, Type, Set, List

_T = TypeVar("_T")

class _Event(Generic[_T]):
    def __init__(self, broadcaster: Callable[..., _T]) -> None:
        self._callbacks: Set[Callable[[_T], None]] = set()
        self.broadcaster = broadcaster

    def subscribe(self, callback: Callable[[_T], None]) -> None:
        """Register the given callable as a subscriber.

        :param callback: a function-like object that will be called
            each time the event is triggered. It must be hashable

    def unsubscribe(self, callback: Callable[[_T], None]) -> None:
        """Unregister the given callable.

        :param callback: the subscriber to unregister
        :raises KeyError: if the given callable in not registered

    def __call__(self, *args: Any, **kwargs: Any) -> _T:
            result = self.broadcaster(*args, **kwargs)
        except TypeError as exception:
            raise TypeError(
                f"{exception}\nIf this is a method and `self` is missing, "
                f"consider adding the `eventclass` decorator on top of the "
                f"declaring class."
            ) from exception
        for callback in self._callbacks:
        return result

def _duplicate_events(obj: object) -> None:
    for attr_name in dir(obj):
        attr = getattr(obj, attr_name)
        if not isinstance(attr, _Event):
        # Creating a new Event specifically for "obj" to
        # avoid side effects on other instances:
        event_for_instance = _Event(attr.broadcaster)
            MethodType(event_for_instance, obj),

def event(broadcaster: Callable[..., _T]) -> _Event[_T]:
    """A decorator that transforms a function into an event.
    Other functions ("subscribers") can subscribe to this event,
    and will be called each time this event is triggered. The
    subscribers must take a single argument that corresponds
    to the output of the event.

    .. warning::
        When decorating a method, the declaring class should be decorated
        with an :func:`eventclass`

    >>> calls: List[str] = []
    >>> @event
    ... def on_greetings() -> str:
    ...     return "Hello"
    >>> def append_target(broadcaster_output: str) -> None:
    ...     calls.append(f"{broadcaster_output} world!")
    >>> on_greetings.subscribe(append_target)
    >>> on_greetings()
    >>> calls
    ['Hello world!']

    :param broadcaster: The function to decorate, to turn it into an event
    :return: An event, which other functions can subscribe to or unsubscribe from
    return _Event(broadcaster)

def eventclass(class_: Type[_T]) -> Type[_T]:
    """A class decorator to enable the decoration of methods with :func:`event`

    >>> calls: List[str] = []
    ... @eventclass
    ... class GUI:
    ...     @event
    ...     def on_button_click(self) -> str:
    ...         return "Clicked"
    >>> def append_target(broadcaster_output: str) -> None:
    ...     calls.append(f"{broadcaster_output} to say hello!")
    >>> gui = GUI()
    >>> gui.on_button_click.subscribe(append_target)
    >>> gui.on_button_click()
    >>> calls
    ['Clicked to say hello!']

    :param class_: The class to decorate
    :return: The corresponding subclass that supports event methods

    class EventClass(class_):  # type: ignore
        def __init__(self, *args: Any, **kwargs: Any) -> None:
            super().__init__(*args, **kwargs)

    return EventClass


from typing import Type, Protocol
from unittest.mock import MagicMock

import pytest

from event import event, eventclass, _Event

def fixture_on_say_hello() -> _Event[str]:
    def on_say_hello() -> str:
        return "Hello world!"

    return on_say_hello

class Greeter(Protocol):
    def on_say_hello_to(self, name: str) -> str: ...

def fixture_greeter() -> Type[Greeter]:
    class _Greeter:
        def on_say_hello_to(self, name: str) -> str:
            return f"Hello {name}!"

    return _Greeter

def test_subscribe_given_function(on_say_hello: _Event[str]) -> None:
    callback = MagicMock()


    assert on_say_hello() == "Hello world!"
    callback.assert_called_once_with("Hello world!")

def test_unsubscribe_given_function(on_say_hello: _Event[str]) -> None:
    callback = MagicMock()


    assert on_say_hello() == "Hello world!"

def test_unsubscribe_given_unregistered_function_raises_key_error(
    on_say_hello: _Event[str],
) -> None:
    callback = MagicMock()

    with pytest.raises(KeyError):

def test_instance_of_decorated_class_does_not_change_of_type() -> None:
    class EventClass: ...

    event_obj = EventClass()

    assert isinstance(event_obj, EventClass)

def test_subscribe_given_method(greeter: Type[Greeter]) -> None:
    callback = MagicMock()
    event_obj = greeter()


    assert event_obj.on_say_hello_to("world") == "Hello world!"
    callback.assert_called_once_with("Hello world!")

def test_subscribe_given_two_instances_keeps_events_isolated(
    greeter: Type[Greeter],
) -> None:
    a = greeter()
    b = greeter()
    callback_a = MagicMock()
    callback_b = MagicMock()


    assert a.on_say_hello_to(name="'a'") == "Hello 'a'!"
    callback_a.assert_called_once_with("Hello 'a'!")

def test_forgetting_event_class_raises_type_error(
    greeter: Type[Greeter],
) -> None:
    event_obj = greeter()
    with pytest.raises(TypeError, match="eventclass"):


1 Answer 1



nit, typo in module name.

... transforms a function into an event

quibble: Maybe "... into an event generator"

modern type annotations

from typing import ... , Set, List

That's a little weird, in the sense of being old.

Prefer set and list in recently written annotations.

LGTM, ship it!


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.