Event library (unreliable outside of CPython)

event.py

# Minimal library to create events

import inspect
from weakref import ref
from weakmethod import WeakMethod

class Event(object):
    def __init__(self, handlers=[]):
        self.handlers = set(handlers)

    def __call__(self, *args, **kwargs):
        lost_references = set()

        for ref_handler in self.handlers:
            try:
                ref_handler()(*args, **kwargs)
            except ReferenceError:
                lost_references.add(handler)

        for lost_ref in lost_references:
            self.handlers.remove(lost_ref)

    def __iadd__(self, handler):
        # bound methods are garbage collected on themselves, not when its bound
        # instance is destroyed.
        # Use WeakMethod to avoid that.
        if inspect.ismethod(handler):
            handler = WeakMethod(handler, callback=self.handlers.remove)
        else:
            handler = ref(handler, callback=self.handlers.remove)

        self.handlers.add(handler)
        return self

test_events.py

from event import Event
import unittest


class TestEvents(unittest.TestCase):
    def setUp(self):
        self.on_test = Event()
        self.called_a = False
        self.called_b = False
        self.called_c = False

    def a(self):
        self.called_a = True

    def b(self):
        self.called_b = True

    def test_event(self):
        # Add event handlers
        self.on_test += self.a
        self.on_test += self.b
        self.assertEqual(len(self.on_test.handlers), 2)

        # Signal the event
        self.on_test()

        self.assertTrue(self.called_a)
        self.assertTrue(self.called_b)
        self.assertFalse(self.called_c)

    def test_class(self):
        # A class which uses the reference
        class MyClass(object):
            def __init__(self, test_object):
                self.test_object = test_object
                self.test_object.on_test += self.c

            def c(self):
                print("c")
                self.test_object.called_c = True
            
        # Add event handlers
        self.on_test += self.a
        self.on_test += self.b
        instance = MyClass(self) # MyClass adds itself to the event
        self.assertEqual(len(self.on_test.handlers), 3)

        # Signal the event
        self.on_test()

        # All event handlers are called
        self.assertTrue(self.called_a)
        self.assertTrue(self.called_b)
        self.assertTrue(self.called_c)

        return instance

    def test_weakref(self):
        instance = self.test_class()
        self.assertEqual(len(self.on_test.handlers), 3)

        # Reset the flags
        self.called_a = False
        self.called_b = False
        self.called_c = False

        # Dispose of instance
        instance = None

        # Signal the event
        self.on_test()

        # The C handler should have been destroyed and thus not called
        self.assertTrue(self.called_a)
        self.assertTrue(self.called_b)
        self.assertFalse(self.called_c)

        # The handler should have been removed
        self.assertEqual(len(self.on_test.handlers), 2)


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

weakmethod.py

import weakref

if hasattr(weakref, "WeakMethod"):
    from weakref import WeakMethod
else:
    from weakref import ref

    class WeakMethod(ref):
        """
        A custom `weakref.ref` subclass which simulates a weak reference to
        a bound method, working around the lifetime problem of bound methods.
        """

        __slots__ = "_func_ref", "_meth_type", "_alive", "__weakref__"

        def __new__(cls, meth, callback=None):
            try:
                obj = meth.__self__
                func = meth.__func__
            except AttributeError:
                raise TypeError("argument should be a bound method, not {}"
                                .format(type(meth)))
            def _cb(arg):
                # The self-weakref trick is needed to avoid creating a reference
                # cycle.
                self = self_wr()
                if self._alive:
                    self._alive = False
                    if callback is not None:
                        callback(self)
            self = ref.__new__(cls, obj, _cb)
            self._func_ref = ref(func, _cb)
            self._meth_type = type(meth)
            self._alive = True
            self_wr = ref(self)
            return self

        def __call__(self):
            obj = super(WeakMethod, self).__call__()
            func = self._func_ref()
            if obj is None or func is None:
                return None
            return self._meth_type(func, obj)

        def __eq__(self, other):
            if isinstance(other, WeakMethod):
                if not self._alive or not other._alive:
                    return self is other
                return ref.__eq__(self, other) and self._func_ref == other._func_ref
            return False

        def __ne__(self, other):
            if isinstance(other, WeakMethod):
                if not self._alive or not other._alive:
                    return self is not other
                return ref.__ne__(self, other) or self._func_ref != other._func_ref
            return True

        __hash__ = ref.__hash__