aboutsummaryrefslogtreecommitdiff
path: root/pyee/cls.py
blob: 21885b49a0c6bcb997fb2224a65d9922f1e017ee (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
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
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