00_Event_Target_System.md

Hunter's Event Target System

Here is a little event system I've been working on for Python. I've been trying to make it typesafe as possible, hence the generics and depending on specific classes for event info. This system has a couple neat features:

1. Automatic event ID creation/decorator based event creation
2. Events call parent events
3. Low overhead (computations on IDs are done once and saved, simple dictionary based listener structure)

Creating and listening to events is simple:

from dataclasses import dataclass
from eventtarget import event, EventTarget

@event
@dataclass
class GreetEvent():
    target: str

target = EventTarget() # I recommend inheriting this class instead, but this works too.
target.subscribe(GreetEvent, lambda event: print(f"Hello, {event.target}!")) 

target.emit(GreetEvent("world")) # Prints "Hello, world!"

eventtarget.py

from abc import ABC, abstractstaticmethod
from collections import defaultdict
from typing import Callable, TypeVar
from itertools import chain
import re


class __Event(ABC):
    """
    !!!!!!!!!!

    DO NOT IMPLEMENT THIS CLASS DIRECTLY! USE THE `@event` DECORATOR!

    !!!!!!!!!!
    """
    @abstractstaticmethod
    def get_id() -> str:
        """
        This event's unique event ID

        Used to determine which listener events are added to
        """
        pass

    @abstractstaticmethod
    def get_ids() -> set[str]:
        """
        A set of event IDs that this event inherits from

        Used to determine which listeners are called when this event is emitted
        """
        pass


E = TypeVar('E', bound=__Event)


class EventTarget():
    def __init__(self):
        self.__listeners = defaultdict(list)

    def subscribe(self, eventType: type[E], listener: Callable[[E], None]):
        """
        Subscribes to an event, running a function when it or any of its
        subclasses are called.
        """
        self.__listeners[eventType.get_id()].append(listener)

    def unsubscribe(self, eventType: type[E], listener: Callable[[E], None]):
        listeners = self.__listeners[eventType.get_id()]
        while listener in listeners:
            listeners.remove(listener)

    def emit(self, event: E):
        """
        Emits an event, running a function on it and all of its base event
        classes. Order is non deterministic (parent events may be called before
        child events, or vice versa).
        """
        listeners = chain(*[self.__listeners[id] for id in event.get_ids()])
        for listener in listeners:
            listener(event)


__upper_camel_pattern = re.compile(r'(?<!^)(?=[A-Z])')


def event(cls: type):
    """
    Implements the event class based on class name and any inherited events.

    Other decorators should be called *below* this decorator (this decorator
    should be called last).
    """
    # create event ID from class name
    event_id = __upper_camel_pattern.sub('_', cls.__name__).lower()

    # find parents that are also events
    event_parents = list(filter(lambda parent: __Event in parent.__mro__, cls.__bases__))

    # generate event ids
    event_ids = set(chain.from_iterable([
        [event_id],
        *[parent.get_ids() for parent in event_parents
            if parent.get_ids() is not None]
    ]))

    # add event parent class if needed
    required_additions = [cls, __Event if len(event_parents) < 1 else None]
    actual_additions = [parent for parent in required_additions
                        if parent is not None]

    # implement class
    class __AnonymousEvent(*actual_additions):
        @staticmethod
        def get_id():
            return event_id

        @staticmethod
        def get_ids():
            return event_ids

    # fix class data
    __AnonymousEvent.__name__ = cls.__name__
    __AnonymousEvent.__qualname__ = cls.__qualname__
    __AnonymousEvent.__module__ = cls.__module__
    __AnonymousEvent.__init__ = cls.__init__

    return __AnonymousEvent

eventtarget.test.py

from dataclasses import dataclass
import unittest
from eventtarget import event, EventTarget
from unittest import TestCase


@event
class TestEvent():
    pass


@event
@dataclass
class TestDataEvent(TestEvent):
    message: str


@event
@dataclass
class TestCountDataEvent(TestEvent):
    count: int


@event
class TestJointDataEvent(TestDataEvent, TestCountDataEvent):
    def __init__(self, message, count):
        TestDataEvent.__init__(self, message)
        TestCountDataEvent.__init__(self, count)


class TestEventTarget(EventTarget):
    def __init__(self):
        super().__init__()


class TestEventTargetCase(TestCase):
    def test_decorated(self):
        event = TestEvent()

        self.assertEqual(event.get_id(), "test_event")
        self.assertSetEqual(event.get_ids(), set(['test_event']))

    def test_data_decorated(self):
        event = TestDataEvent("hello")

        self.assertEqual(event.get_id(), "test_data_event")
        self.assertSetEqual(event.get_ids(), set([
            'test_data_event',
            'test_event'
        ]))
        self.assertEqual(event.message, "hello")

    def test_multiple_decorated(self):
        event = TestJointDataEvent("hello", 5)

        self.assertEqual(event.get_id(), "test_joint_data_event")
        self.assertSetEqual(event.get_ids(), set([
            'test_joint_data_event',
            'test_count_data_event',
            'test_data_event',
            'test_event'
        ]))

    def test_listen_emit(self):
        target = TestEventTarget()
        called = False

        def fn(_):
            nonlocal called
            called = True

        target.subscribe(TestEvent, fn)
        event = TestEvent()
        target.emit(event)
        self.assertTrue(called, "should have emitted the event")

    def test_listen_child(self):
        target = TestEventTarget()
        called = False
        data = None

        def fn(_: TestEvent):
            nonlocal called
            called = True

        def child_fn(event: TestDataEvent):
            nonlocal data
            data = event.message

        target.subscribe(TestEvent, fn)
        target.subscribe(TestDataEvent, child_fn)

        event = TestDataEvent("hello")
        target.emit(event)

        self.assertTrue(called, "should have emitted the event")
        self.assertEqual(data, "hello", "should have attached the data")

    def test_unsubscribe(self):
        target = TestEventTarget()
        call_count = 0

        def fn(_: TestEvent):
            nonlocal call_count
            call_count += 1

        target.subscribe(TestEvent, fn)

        event = TestEvent()
        target.emit(event)

        self.assertEqual(call_count, 1, "should have listened to the event")

        target.unsubscribe(TestEvent, fn)

        self.assertEqual(call_count, 1, "should not have listened to the event")


if __name__ == "__main__":
    unittest.main()