1from dataclasses import dataclass 2from functools import wraps 3from typing import Callable, List, Type, TypeVar 4 5from pyee import EventEmitter 6 7 8@dataclass 9class Handler: 10 event: str 11 method: Callable 12 13 14class Handlers: 15 def __init__(self): 16 self._handlers: List[Handler] = [] 17 18 def append(self, handler): 19 self._handlers.append(handler) 20 21 def __iter__(self): 22 return iter(self._handlers) 23 24 def reset(self): 25 self._handlers = [] 26 27 28_handlers = Handlers() 29 30 31def on(event: str) -> Callable[[Callable], Callable]: 32 """ 33 Register an event handler on an evented class. See the ``evented`` class 34 decorator for a full example. 35 """ 36 37 def decorator(method: Callable) -> Callable: 38 _handlers.append(Handler(event=event, method=method)) 39 return method 40 41 return decorator 42 43 44def _bind(self, method): 45 @wraps(method) 46 def bound(*args, **kwargs): 47 return method(self, *args, **kwargs) 48 49 return bound 50 51 52Cls = TypeVar(name="Cls", bound=Type) 53 54 55def evented(cls: Cls) -> Cls: 56 """ 57 Configure an evented class. 58 59 Evented classes are classes which use an EventEmitter to call instance 60 methods during runtime. To achieve this without this helper, you would 61 instantiate an ``EventEmitter`` in the ``__init__`` method and then call 62 ``event_emitter.on`` for every method on ``self``. 63 64 This decorator and the ``on`` function help make things look a little nicer 65 by defining the event handler on the method in the class and then adding 66 the ``__init__`` hook in a wrapper:: 67 68 from pyee.cls import evented, on 69 70 @evented 71 class Evented: 72 @on("event") 73 def event_handler(self, *args, **kwargs): 74 print(self, args, kwargs) 75 76 evented_obj = Evented() 77 78 evented_obj.event_emitter.emit( 79 "event", "hello world", numbers=[1, 2, 3] 80 ) 81 82 The ``__init__`` wrapper will create a ``self.event_emitter: EventEmitter`` 83 automatically but you can also define your own event_emitter inside your 84 class's unwrapped ``__init__`` method. For example, to use this 85 decorator with a ``TwistedEventEmitter``:: 86 87 @evented 88 class Evented: 89 def __init__(self): 90 self.event_emitter = TwistedEventEmitter() 91 92 @on("event") 93 async def event_handler(self, *args, **kwargs): 94 await self.some_async_action(*args, **kwargs) 95 """ 96 handlers: List[Handler] = list(_handlers) 97 _handlers.reset() 98 99 og_init: Callable = cls.__init__ 100 101 @wraps(cls.__init__) 102 def init(self, *args, **kwargs): 103 og_init(self, *args, **kwargs) 104 if not hasattr(self, "event_emitter"): 105 self.event_emitter = EventEmitter() 106 107 for h in handlers: 108 self.event_emitter.on(h.event, _bind(self, h.method)) 109 110 cls.__init__ = init 111 112 return cls 113