Merge pull request #17 from strohganoff/refactor/event-handler-catalogs

Refactor/event handler catalogs

Author: strohganoff

streamdeck/__init__.py

@@ -1,11 +1,11 @@
 from . import (
-    actions,
     command_sender,
     manager,
     models,
     utils,
     websocket,
 )
+from .event_handlers import actions
 
 
 __all__ = [

tests/actions/__init__.py streamdeck/event_handlers/__init__.pytests/actions/__init__.py renamed to streamdeck/event_handlers/__init__.py

@@ -6,36 +6,32 @@
 from logging import getLogger
 from typing import TYPE_CHECKING, cast
 
+from streamdeck.event_handlers.protocol import (
+    EventHandlerFunc,
+    EventModel_contra,
+    InjectableParams,
+    SupportsEventHandlers,
+)
+
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Generator
-    from typing import Protocol
-
-    from typing_extensions import ParamSpec, TypeVar
 
-    from streamdeck.models.events import EventBase
     from streamdeck.types import ActionUUIDStr, EventNameStr
 
 
 
-    EventModel_contra = TypeVar("EventModel_contra", bound=EventBase, default=EventBase, contravariant=True)
-    InjectableParams = ParamSpec("InjectableParams", default=...)
-
-    class EventHandlerFunc(Protocol[EventModel_contra, InjectableParams]):
-        """Protocol for an event handler function that takes an event (of subtype of EventBase) and other parameters that are injectable."""
-        def __call__(self, event_data: EventModel_contra, *args: InjectableParams.args, **kwargs: InjectableParams.kwargs) -> None: ...
-
-
-
 logger = getLogger("streamdeck.actions")
 
 
-class ActionBase(ABC):
+class ActionBase(ABC, SupportsEventHandlers):
     """Base class for all actions."""
+    _events: dict[EventNameStr, set[EventHandlerFunc]]
+    """Dictionary mapping event names to sets of event handler functions."""
 
     def __init__(self) -> None:
         """Initialize an Action instance."""
-        self._events: dict[EventNameStr, set[EventHandlerFunc]] = defaultdict(set)
+        self._events = defaultdict(set)
 
     def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[EventModel_contra, InjectableParams]], EventHandlerFunc[EventModel_contra, InjectableParams]]:
         """Register an event handler for a specific event.
@@ -88,6 +84,8 @@ class GlobalAction(ActionBase):
 
 class Action(ActionBase):
     """Represents an action that can be performed for a specific action, with event handlers for specific event types."""
+    uuid: ActionUUIDStr
+    """The unique identifier for the action."""
 
     def __init__(self, uuid: ActionUUIDStr) -> None:
         """Initialize an Action instance.
@@ -104,37 +102,3 @@ def name(self) -> str:
         return self.uuid.split(".")[-1]
 
 
-class ActionRegistry:
-    """Manages the registration and retrieval of actions and their event handlers."""
-
-    def __init__(self) -> None:
-        """Initialize an ActionRegistry instance."""
-        self._plugin_actions: list[ActionBase] = []
-
-    def register(self, action: ActionBase) -> None:
-        """Register an action with the registry.
-
-        Args:
-            action (Action): The action to register.
-        """
-        self._plugin_actions.append(action)
-
-    def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: ActionUUIDStr | None = None) -> Generator[EventHandlerFunc, None, None]:
-        """Get all event handlers for a specific event from all registered actions.
-
-        Args:
-            event_name (EventName): The name of the event to retrieve handlers for.
-            event_action_uuid (str | None): The action UUID to get handlers for.
-                If None (i.e., the event is not action-specific), get all handlers for the event.
-
-        Yields:
-            EventHandlerFunc: The event handler functions for the specified event.
-        """
-        for action in self._plugin_actions:
-            # If the event is action-specific (i.e is not a GlobalAction and has a UUID attribute),
-            # only get handlers for that action, as we don't want to trigger
-            # and pass this event to handlers for other actions.
-            if event_action_uuid is not None and (isinstance(action, Action) and action.uuid != event_action_uuid):
-                continue
-
-            yield from action.get_event_handlers(event_name)

streamdeck/actions.py streamdeck/event_handlers/actions.pystreamdeck/actions.py renamed to streamdeck/event_handlers/actions.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import inspect
+from collections.abc import Iterable
+from typing import TYPE_CHECKING, Protocol, runtime_checkable
+
+from typing_extensions import ParamSpec, TypeGuard, TypeVar  # noqa: UP035
+
+from streamdeck.command_sender import StreamDeckCommandSender
+from streamdeck.models.events import EventBase
+
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
+    from streamdeck.types import EventNameStr
+
+
+
+EventModel_contra = TypeVar("EventModel_contra", bound=EventBase, default=EventBase, contravariant=True)
+InjectableParams = ParamSpec("InjectableParams", default=...)
+
+
+class EventHandlerFunc(Protocol[EventModel_contra, InjectableParams]):
+    """Protocol for an event handler function that takes an event (of subtype of EventBase) and other parameters that are injectable."""
+    def __call__(self, event_data: EventModel_contra, *args: InjectableParams.args, **kwargs: InjectableParams.kwargs) -> None: ...
+
+
+BindableEventHandlerFunc = EventHandlerFunc[EventModel_contra, tuple[StreamDeckCommandSender]]  # type: ignore[misc]
+"""Type alias for a bindable event handler function that takes an event (of subtype of EventBase) and a command_sender parameter that is to be injected."""
+BoundEventHandlerFunc = EventHandlerFunc[EventModel_contra]
+"""Type alias for a bound event handler function that takes an event (of subtype of EventBase) and no other parameters.
+
+Typically used for event handlers that have already had parameters injected.
+"""
+
+
+@runtime_checkable
+class SupportsEventHandlers(Protocol):
+    """Protocol for a class that holds event-specific handler functions which can be pulled out by event name.
+
+    Implementing classes should handle defining the registration of event handlers and ensuring
+    that they can be retrieved efficiently.
+    """
+
+    def get_event_handlers(self, event_name: EventNameStr, /) -> Iterable[EventHandlerFunc]:
+        """Get all event handlers for a specific event.
+
+        Args:
+            event_name (str): The name of the event to get handlers for.
+
+        Returns:
+            Iterable[EventHandlerFunc]: The event handler functions for the specified event.
+        """
+        ...
+
+    def get_registered_event_names(self) -> list[EventNameStr]:
+        """Get all event names for which handlers are registered.
+
+        Returns:
+            list[str]: A list of event names for which handlers are registered.
+        """
+        ...
+
+
+# def is_bindable_handler(handler: EventHandlerFunc[EventModel_contra, InjectableParams]) -> TypeGuard[BindableEventHandlerFunc[EventModel_contra]]:
+def is_bindable_handler(handler: EventHandlerFunc[EventModel_contra, InjectableParams]) -> bool:
+    """Check if the handler is prebound with the `command_sender` parameter."""
+    # Check dynamically if the `command_sender`'s name is in the handler's arguments.
+    return "command_sender" in inspect.signature(handler).parameters
+
+
+def is_not_bindable_handler(handler: EventHandlerFunc[EventModel_contra, InjectableParams]) -> TypeGuard[BoundEventHandlerFunc[EventModel_contra]]:
+    """Check if the handler only accepts the event_data parameter.
+
+    If this function returns False after the is_bindable_handler check is True, then the function has invalid parameters, and will subsequently need to be handled in the calling code.
+    """
+    handler_params = inspect.signature(handler).parameters
+    return len(handler_params) == 1 and "event_data" in handler_params
+
+
+
+
+
+
+
+

streamdeck/event_handlers/protocol.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from streamdeck.event_handlers.actions import Action
+
+
+if TYPE_CHECKING:
+    from collections.abc import Generator
+
+    from streamdeck.event_handlers.protocol import EventHandlerFunc, SupportsEventHandlers
+    from streamdeck.types import ActionUUIDStr, EventNameStr
+
+
+class HandlersRegistry:
+    """Manages the registration and retrieval of event handler catalogs and their event handlers."""
+    _plugin_event_handler_catalogs: list[SupportsEventHandlers]
+    """List of registered actions and other event handler catalogs."""
+
+    def __init__(self) -> None:
+        """Initialize a HandlersRegistry instance."""
+        self._plugin_event_handler_catalogs = []
+
+    def register(self, catalog: SupportsEventHandlers) -> None:
+        """Register an event handler catalog with the registry.
+
+        Args:
+            catalog (SupportsEventHandlers): The event handler catalog to register.
+        """
+        self._plugin_event_handler_catalogs.append(catalog)
+
+    def get_event_handlers(self, event_name: EventNameStr, event_action_uuid: ActionUUIDStr | None = None) -> Generator[EventHandlerFunc, None, None]:
+        """Get all event handlers for a specific event from all registered event handler catalogs.
+
+        Args:
+            event_name (EventName): The name of the event to retrieve handlers for.
+            event_action_uuid (str | None): The action UUID to get handlers for.
+                If None (i.e., the event is not action-specific), get all handlers for the event.
+
+        Yields:
+            EventHandlerFunc: The event handler functions for the specified event.
+        """
+        for catalog in self._plugin_event_handler_catalogs:
+            # If the event is action-specific (i.e is not a GlobalAction and has a UUID attribute),
+            # only get handlers for that action, as we don't want to trigger
+            # and pass this event to handlers for other actions.
+            if event_action_uuid is not None and (isinstance(catalog, Action) and catalog.uuid != event_action_uuid):
+                continue
+
+            yield from catalog.get_event_handlers(event_name)