diff --git a/docs/index.rst b/docs/index.rst index 5a6441b..c410f5c 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -43,6 +43,10 @@ API Docs: .. autofunction:: pyee.uplift.uplift +.. autofunction:: pyee.cls.on + +.. autofunction:: pyee.cls.evented + Some Links ========== diff --git a/pyee/cls.py b/pyee/cls.py new file mode 100644 index 0000000..21885b4 --- /dev/null +++ b/pyee/cls.py @@ -0,0 +1,112 @@ +from dataclasses import dataclass +from functools import wraps +from typing import Callable, List, Type, TypeVar + +from pyee import EventEmitter + + +@dataclass +class Handler: + event: str + method: Callable + + +class Handlers: + def __init__(self): + self._handlers: List[Handler] = [] + + def append(self, handler): + self._handlers.append(handler) + + def __iter__(self): + return iter(self._handlers) + + def reset(self): + self._handlers = [] + + +_handlers = Handlers() + + +def on(event: str) -> Callable[[Callable], Callable]: + """ + Register an event handler on an evented class. See the ``evented`` class + decorator for a full example. + """ + + def decorator(method: Callable) -> Callable: + _handlers.append(Handler(event=event, method=method)) + return method + + return decorator + + +def _bind(self, method): + @wraps(method) + def bound(*args, **kwargs): + return method(self, *args, **kwargs) + + return bound + + +Cls = TypeVar(name="Cls", bound=Type) + + +def evented(cls: Cls) -> Cls: + """ + Configure an evented class. + + Evented classes are classes which use an EventEmitter to call instance + methods during runtime. To achieve this without this helper, you would + instantiate an ``EventEmitter`` in the ``__init__`` method and then call + ``event_emitter.on`` for every method on ``self``. + + This decorator and the ``on`` function help make things look a little nicer + by defining the event handler on the method in the class and then adding + the ``__init__`` hook in a wrapper:: + + from pyee.cls import evented, on + + @evented + class Evented: + @on("event") + def event_handler(self, *args, **kwargs): + print(self, args, kwargs) + + evented_obj = Evented() + + evented_obj.event_emitter.emit( + "event", "hello world", numbers=[1, 2, 3] + ) + + The ``__init__`` wrapper will create a ``self.event_emitter: EventEmitter`` + automatically but you can also define your own event_emitter inside your + class's unwrapped ``__init__`` method. For example, to use this + decorator with a ``TwistedEventEmitter``:: + + @evented + class Evented: + def __init__(self): + self.event_emitter = TwistedEventEmitter() + + @on("event") + async def event_handler(self, *args, **kwargs): + await self.some_async_action(*args, **kwargs) + """ + handlers: List[Handler] = list(_handlers) + _handlers.reset() + + og_init: Callable = cls.__init__ + + @wraps(cls.__init__) + def init(self, *args, **kwargs): + og_init(self, *args, **kwargs) + if not hasattr(self, "event_emitter"): + self.event_emitter = EventEmitter() + + for h in handlers: + self.event_emitter.on(h.event, _bind(self, h.method)) + + cls.__init__ = init + + return cls diff --git a/tests/test_cls.py b/tests/test_cls.py new file mode 100644 index 0000000..d7ca3ec --- /dev/null +++ b/tests/test_cls.py @@ -0,0 +1,47 @@ +# -*- coding: utf-8 -*- +from mock import Mock +import pytest + +from pyee import EventEmitter +from pyee.cls import evented, on + + +@evented +class EventedFixture: + def __init__(self): + self.call_me = Mock() + + @on("event") + def event_handler(self, *args, **kwargs): + self.call_me(self, *args, **kwargs) + + +_custom_event_emitter = EventEmitter() + + +@evented +class CustomEmitterFixture: + def __init__(self): + self.call_me = Mock() + self.event_emitter = _custom_event_emitter + + @on("event") + def event_handler(self, *args, **kwargs): + self.call_me(self, *args, **kwargs) + + +class InheritedFixture(EventedFixture): + pass + + +@pytest.mark.parametrize( + "cls", [EventedFixture, CustomEmitterFixture, InheritedFixture] +) +def test_evented_decorator(cls): + inst = cls() + + inst.event_emitter.emit("event", "emitter is emitted!") + + inst.call_me.assert_called_once_with(inst, "emitter is emitted!") + + _custom_event_emitter.remove_all_listeners()