Update EventHandlerFunc protocols. Delete streamdeck/types.py

Author: strohganoff

streamdeck/__init__.py

@@ -3,7 +3,6 @@
     command_sender,
     manager,
     models,
-    types,
     utils,
     websocket,
 )
@@ -14,7 +13,6 @@
     "command_sender",
     "manager",
     "models",
-    "types",
     "utils",
     "websocket",
 ]

streamdeck/actions.py

@@ -9,10 +9,27 @@
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Generator
+    from typing import Protocol
+
+    from typing_extensions import ParamSpec, TypeAlias, TypeVar  # noqa: UP035
 
     from streamdeck.models.events import EventBase
-    from streamdeck.models.events.base import LiteralStrGenericAlias
-    from streamdeck.types import BaseEventHandlerFunc, EventHandlerFunc, EventNameStr, TEvent_contra
+
+
+    EventNameStr: TypeAlias = str  # noqa: UP040
+    """Type alias for the event name string.
+
+    We don't define literal string values here, as the list of available event names can be added to dynamically.
+    """
+
+
+    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")
@@ -27,9 +44,9 @@ def __init__(self) -> None:
         Args:
             uuid (str): The unique identifier for the action.
         """
-        self._events: dict[EventNameStr, set[BaseEventHandlerFunc]] = defaultdict(set)
+        self._events: dict[EventNameStr, set[EventHandlerFunc]] = defaultdict(set)
 
-    def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[TEvent_contra] | BaseEventHandlerFunc], EventHandlerFunc[TEvent_contra] | BaseEventHandlerFunc]:
+    def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[EventModel_contra, InjectableParams]], EventHandlerFunc[EventModel_contra, InjectableParams]]:
         """Register an event handler for a specific event.
 
         Args:
@@ -41,15 +58,14 @@ def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[TEvent_c
         Raises:
             KeyError: If the provided event name is not available.
         """
-        def _wrapper(func: EventHandlerFunc[TEvent_contra]) -> EventHandlerFunc[TEvent_contra]:
-            # Cast to BaseEventHandlerFunc so that the storage type is consistent.
-            self._events[event_name].add(cast("BaseEventHandlerFunc", func))
-
+        def _wrapper(func: EventHandlerFunc[EventModel_contra, InjectableParams]) -> EventHandlerFunc[EventModel_contra, InjectableParams]:
+            # Cast to EventHandlerFunc with default type arguments so that the storage type is consistent.
+            self._events[event_name].add(cast("EventHandlerFunc", func))
             return func
 
         return _wrapper
 
-    def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHandlerFunc[EventBase], None, None]:
+    def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHandlerFunc, None, None]:
         """Get all event handlers for a specific event.
 
         Args:
@@ -112,7 +128,7 @@ def register(self, action: ActionBase) -> None:
         """
         self._plugin_actions.append(action)
 
-    def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: str | None = None) -> Generator[EventHandlerFunc[EventBase], None, None]:
+    def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: str | None = None) -> Generator[EventHandlerFunc, None, None]:
         """Get all event handlers for a specific event from all registered actions.
 
         Args:

streamdeck/manager.py

@@ -1,22 +1,18 @@
 from __future__ import annotations
 
 import functools
+import inspect
 from logging import getLogger
 from typing import TYPE_CHECKING
 
 from pydantic import ValidationError
+from typing_extensions import TypeGuard  # noqa: UP035
 
 from streamdeck.actions import Action, ActionBase, ActionRegistry
 from streamdeck.command_sender import StreamDeckCommandSender
 from streamdeck.event_listener import EventListener, EventListenerManager
 from streamdeck.models.events.adapter import EventAdapter
 from streamdeck.models.events.common import ContextualEventMixin
-from streamdeck.types import (
-    EventHandlerBasicFunc,
-    EventHandlerFunc,
-    TEvent_contra,
-    is_bindable_handler,
-)
 from streamdeck.utils.logging import configure_streamdeck_logger
 from streamdeck.websocket import WebSocketClient
 
@@ -25,13 +21,42 @@
     from collections.abc import Generator
     from typing import Any, Literal
 
+    from streamdeck.actions import (
+        EventHandlerFunc,
+        EventModel_contra,
+        InjectableParams,
+    )
     from streamdeck.models.events import EventBase
 
 
+    BindableEventHandlerFunc = EventHandlerFunc[EventModel_contra, [StreamDeckCommandSender]]
+    """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.
+    """
+
+
+
 # TODO: Fix this up to push to a log in the apropos directory and filename.
 logger = getLogger("streamdeck.manager")
 
 
+def is_bindable_handler(handler: EventHandlerFunc[EventModel_contra, InjectableParams]) -> TypeGuard[BindableEventHandlerFunc[EventModel_contra]]:
+    """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
+
 
 class PluginManager:
     """Manages plugin actions and communicates with a WebSocket server to handle events."""
@@ -105,7 +130,7 @@ def register_event_listener(self, listener: EventListener) -> None:
         for event_model in listener.event_models:
             self._event_adapter.add_model(event_model)
 
-    def _inject_command_sender(self, handler: EventHandlerFunc[TEvent_contra], command_sender: StreamDeckCommandSender) -> EventHandlerBasicFunc[TEvent_contra]:
+    def _inject_command_sender(self, handler: EventHandlerFunc[EventModel_contra, InjectableParams], command_sender: StreamDeckCommandSender) -> BoundEventHandlerFunc[EventModel_contra]:
         """Inject command_sender into handler if it accepts it as a parameter.
 
         Args:
@@ -116,8 +141,13 @@ def _inject_command_sender(self, handler: EventHandlerFunc[TEvent_contra], comma
             The handler with command_sender injected if needed
         """
         if is_bindable_handler(handler):
+            # If the handler accepts command_sender, inject it and return the handler.
             return functools.partial(handler, command_sender=command_sender)
 
+        if not is_not_bindable_handler(handler):
+            # If the handler is neither bindable nor not bindable, raise an error.
+            raise TypeError(f"Invalid event handler function signature: {handler}")  # noqa: TRY003, EM102
+
         return handler
 
     def _stream_event_data(self) -> Generator[EventBase, None, None]:

streamdeck/types.py

@@ -12,9 +12,8 @@
     from typing import Any, ClassVar
 
 
-class DummyEvent(EventBase):
+class DummyEvent(EventBase["dummy"]):
     """A dummy event for testing purposes."""
-    event: Literal["dummy"]  # type: ignore[assignment]
     something: int
 
 

tests/data/dummy_event_listener.py

@@ -11,17 +11,14 @@
 if TYPE_CHECKING:
     from functools import partial
 
+    from streamdeck.actions import EventHandlerFunc
     from streamdeck.command_sender import StreamDeckCommandSender
     from streamdeck.manager import PluginManager
     from streamdeck.models import events
-    from streamdeck.types import (
-        EventHandlerBasicFunc,
-        EventHandlerFunc,
-    )
 
 
 
-def create_event_handler(include_command_sender_param: bool = False) -> EventHandlerFunc[events.EventBase]:
+def create_event_handler(include_command_sender_param: bool = False) -> EventHandlerFunc:
     """Create a dummy event handler function that matches the EventHandlerFunc TypeAlias.
 
     Args:
@@ -45,7 +42,7 @@ def dummy_handler_with_cmd_sender(event_data: events.EventBase, command_sender:
 @pytest.fixture(params=[True, False])
 def mock_event_handler(request: pytest.FixtureRequest) -> Mock:
     include_command_sender_param: bool = request.param
-    dummy_handler: EventHandlerFunc[events.EventBase] = create_event_handler(include_command_sender_param)
+    dummy_handler: EventHandlerFunc = create_event_handler(include_command_sender_param)
 
     return create_autospec(dummy_handler, spec_set=True)
 
@@ -57,7 +54,7 @@ def test_inject_command_sender_func(
 ) -> None:
     """Test that the command_sender is injected into the handler."""
     mock_command_sender = Mock()
-    result_handler: EventHandlerFunc [events.EventBase]= plugin_manager._inject_command_sender(mock_event_handler, mock_command_sender)
+    result_handler: EventHandlerFunc = plugin_manager._inject_command_sender(mock_event_handler, mock_command_sender)
 
     resulting_handler_params = inspect.signature(result_handler).parameters
 
@@ -78,7 +75,7 @@ def test_inject_command_sender_func(
 
 @pytest.mark.usefixtures("patch_websocket_client")
 def test_run_manager_events_handled_with_correct_params(
-    mock_event_listener_manager_with_fake_events: tuple[Mock, list[events.EventBase[LiteralStrGenericAlias]]],
+    mock_event_listener_manager_with_fake_events: tuple[Mock, list[events.EventBase]],
     plugin_manager: PluginManager,
     mock_command_sender: Mock,
 ) -> None: