make an EventHandlerType protocol

Author: rmorshea

src/idom/core/events.py

@@ -2,32 +2,21 @@
 Events
 """
 
+from __future__ import annotations
+
 import asyncio
-from typing import (
-    Any,
-    Callable,
-    Coroutine,
-    Dict,
-    Iterator,
-    List,
-    Mapping,
-    Optional,
-    Union,
-)
-from uuid import uuid4
+from typing import Any, Callable, Iterator, List, Mapping, Optional, Sequence
 
 from anyio import create_task_group
 
-
-EventsMapping = Union[Dict[str, Union["Callable[..., Any]", "EventHandler"]], "Events"]
+from idom.core.proto import EventHandlerDict, EventHandlerFunc, EventHandlerType
 
 
 def event(
-    function: Optional[Callable[..., Any]] = None,
     stop_propagation: bool = False,
     prevent_default: bool = False,
-) -> Union["EventHandler", Callable[[Callable[..., Any]], "EventHandler"]]:
-    """Create an event handler function with extra functionality.
+) -> Callable[[Callable[..., Any]], EventHandler]:
+    """A decorator for constructing an :class:`EventHandler`.
 
     While you're always free to add callbacks by assigning them to an element's attributes
 
@@ -39,23 +28,78 @@ def event(
     from taking place, or stoping the event from propagating up the DOM. This decorator
     allows you to add that functionality to your callbacks.
 
+    .. code-block:: python
+
+        @event(stop_propagation=True, prevent_default=True)
+        def my_callback(*data):
+            ...
+
+        element = idom.html.button({"onClick": my_callback})
+
     Parameters:
         function:
-            A callback responsible for handling the event.
+            A function or coroutine responsible for handling the event.
         stop_propagation:
             Block the event from propagating further up the DOM.
         prevent_default:
             Stops the default actional associate with the event from taking place.
     """
-    handler = EventHandler(stop_propagation, prevent_default)
-    if function is not None:
-        handler.add(function)
-        return handler
-    else:
-        return handler.add
 
+    def setup(function: Callable[..., Any]) -> EventHandler:
+        return EventHandler(
+            to_event_handler_function(function),
+            stop_propagation,
+            prevent_default,
+        )
+
+    return setup
+
+
+class EventHandler:
+    """Turn a function or coroutine into an event handler
+
+    Parameters:
+        function:
+            The function or coroutine which handles the event.
+        stop_propagation:
+            Block the event from propagating further up the DOM.
+        prevent_default:
+            Stops the default action associate with the event from taking place.
+        target:
+            A unique identifier for this event handler (auto-generated by default)
+    """
 
-class Events(Mapping[str, "EventHandler"]):
+    __slots__ = (
+        "__weakref__",
+        "function",
+        "prevent_default",
+        "stop_propagation",
+        "target",
+    )
+
+    def __init__(
+        self,
+        function: EventHandlerFunc,
+        stop_propagation: bool = False,
+        prevent_default: bool = False,
+        target: Optional[str] = None,
+    ) -> None:
+        self.function = function
+        self.prevent_default = prevent_default
+        self.stop_propagation = stop_propagation
+        self.target = target
+
+    def __repr__(self) -> str:
+        public_names = [name for name in self.__slots__ if not name.startswith("_")]
+        items = ", ".join([f"{n}={getattr(self, n)!r}" for n in public_names])
+        return f"{type(self).__name__}({items})"
+
+
+async def _no_op(data: List[Any]) -> None:
+    return None
+
+
+class Events(Mapping[str, EventHandler]):
     """A container for event handlers.
 
     Assign this object to the ``"eventHandlers"`` field of an element model.
@@ -64,10 +108,13 @@ class Events(Mapping[str, "EventHandler"]):
     __slots__ = "_handlers"
 
     def __init__(self) -> None:
-        self._handlers: Dict[str, EventHandler] = {}
+        self._handlers: EventHandlerDict = {}
 
     def on(
-        self, event: str, stop_propagation: bool = False, prevent_default: bool = False
+        self,
+        event: str,
+        stop_propagation: Optional[bool] = None,
+        prevent_default: Optional[bool] = None,
     ) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
         """A decorator for adding an event handler.
 
@@ -105,13 +152,46 @@ def handler(event):
             event = "on" + event[:1].upper() + event[1:]
 
         if event not in self._handlers:
-            handler = EventHandler(stop_propagation, prevent_default)
-            self._handlers[event] = handler
-        else:
-            handler = self._handlers[event]
+            # do this so it's possible to stop event propagation or default behavior
+            # without making the user have to pass a no op event handler themselves
+            self._handlers[event] = EventHandler(
+                _no_op,
+                stop_propagation,
+                prevent_default,
+            )
 
         def setup(function: Callable[..., Any]) -> Callable[..., Any]:
-            handler.add(function)
+            old_handler = self._handlers[event]
+
+            if old_handler.function is _no_op:
+                return EventHandler(
+                    to_event_handler_function(function),
+                    bool(stop_propagation),
+                    bool(prevent_default),
+                )
+
+            new_stop_propagation = (
+                old_handler.stop_propagation
+                if stop_propagation is None
+                else stop_propagation
+            )
+            new_prevent_default = (
+                old_handler.prevent_default
+                if prevent_default is None
+                else prevent_default
+            )
+
+            self._handlers[event] = merge_event_handlers(
+                [
+                    old_handler,
+                    EventHandler(
+                        to_event_handler_function(function),
+                        new_stop_propagation,
+                        new_prevent_default,
+                    ),
+                ]
+            )
+
             return function
 
         return setup
@@ -125,91 +205,79 @@ def __len__(self) -> int:
     def __iter__(self) -> Iterator[str]:
         return iter(self._handlers)
 
-    def __getitem__(self, key: str) -> "EventHandler":
+    def __getitem__(self, key: str) -> EventHandler:
         return self._handlers[key]
 
     def __repr__(self) -> str:  # pragma: no cover
         return repr(self._handlers)
 
 
-class EventHandler:
-    """An object which defines an event handler.
-
-    The event handler object acts like a coroutine when called.
+def to_event_handler_function(function: Callable[..., Any]) -> EventHandlerFunc:
+    """Make a :data:`~idom.core.proto.EventHandlerFunc` from a function or coroutine
 
     Parameters:
-        stop_propagation:
-            Block the event from propagating further up the DOM.
-        prevent_default:
-            Stops the default action associate with the event from taking place.
+        function:
+            A function or coroutine accepting a number of positional arguments.
     """
+    if asyncio.iscoroutinefunction(function):
+        return lambda data: function(*data)
+    else:
 
-    __slots__ = (
-        "__weakref__",
-        "_coro_handlers",
-        "_func_handlers",
-        "prevent_default",
-        "stop_propagation",
-        "target",
-    )
+        async def wrapper(data: List[Any]) -> None:
+            return function(*data)
 
-    def __init__(
-        self,
-        stop_propagation: bool = False,
-        prevent_default: bool = False,
-    ) -> None:
-        self._coro_handlers: List[Callable[..., Coroutine[Any, Any, Any]]] = []
-        self._func_handlers: List[Callable[..., Any]] = []
-        self.prevent_default = prevent_default
-        self.stop_propagation = stop_propagation
-        self.target = uuid4().hex
+        return wrapper
 
-    def add(self, function: Callable[..., Any]) -> "EventHandler":
-        """Add a callback function or coroutine to the event handler.
 
-        Parameters:
-            function:
-                The event handler function accepting parameters sent by the client.
-                Typically this is a single ``event`` parameter that is a dictionary.
-        """
-        if asyncio.iscoroutinefunction(function):
-            self._coro_handlers.append(function)
-        else:
-            self._func_handlers.append(function)
-        return self
+def merge_event_handlers(event_handlers: Sequence[EventHandlerType]) -> EventHandler:
+    """Merge multiple event handlers into one
 
-    def remove(self, function: Callable[..., Any]) -> None:
-        """Remove the given function or coroutine from this event handler.
+    Raises a ValueError if any handlers have conflicting
+    :attr:`~idom.core.proto.EventHandlerType.stop_propagation` or
+    :attr:`~idom.core.proto.EventHandlerType.prevent_default` attributes.
+    """
+    if not event_handlers:
+        raise ValueError("No event handlers to merge")
+    elif len(event_handlers) == 1:
+        return event_handlers[0]
+
+    first_handler = event_handlers[0]
+
+    stop_propagation = first_handler.stop_propagation
+    prevent_default = first_handler.prevent_default
+    target = first_handler.target
+
+    for handler in event_handlers:
+        if (
+            handler.stop_propagation != stop_propagation
+            or handler.prevent_default != prevent_default
+            or handler.target != target
+        ):
+            raise ValueError(
+                "Cannot merge handlers - "
+                "'stop_propagation', 'prevent_default' or 'target' mistmatch."
+            )
+
+    return EventHandler(
+        merge_event_handler_funcs([h.function for h in event_handlers]),
+        stop_propagation,
+        prevent_default,
+        target,
+    )
 
-        Raises:
-            ValueError: if not found
-        """
-        if asyncio.iscoroutinefunction(function):
-            self._coro_handlers.remove(function)
-        else:
-            self._func_handlers.remove(function)
-
-    def clear(self) -> None:
-        """Remove all functions and coroutines from this event handler"""
-        self._coro_handlers.clear()
-        self._func_handlers.clear()
-
-    async def __call__(self, data: List[Any]) -> Any:
-        """Trigger all callbacks in the event handler."""
-        if self._coro_handlers:
-            async with create_task_group() as group:
-                for handler in self._coro_handlers:
-                    group.start_soon(handler, *data)
-        for handler in self._func_handlers:
-            handler(*data)
-
-    def __contains__(self, function: Any) -> bool:
-        if asyncio.iscoroutinefunction(function):
-            return function in self._coro_handlers
-        else:
-            return function in self._func_handlers
 
-    def __repr__(self) -> str:
-        public_names = [name for name in self.__slots__ if not name.startswith("_")]
-        items = ", ".join([f"{n}={getattr(self, n)!r}" for n in public_names])
-        return f"{type(self).__name__}({items})"
+def merge_event_handler_funcs(
+    functions: Sequence[EventHandlerFunc],
+) -> EventHandlerFunc:
+    """Make one event handler function from many"""
+    if not functions:
+        raise ValueError("No handler functions to merge")
+    elif len(functions) == 1:
+        return functions[0]
+
+    async def await_all_event_handlers(data: List[Any]) -> None:
+        async with create_task_group() as group:
+            for func in functions:
+                group.start_soon(func, data)
+
+    return await_all_event_handlers