From c2e469c41b7fe1acdd7b99399e6eb50a9bdb3311 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Wed, 9 Apr 2025 18:15:59 -0600 Subject: [PATCH 1/5] Minor cleanup and giving EventBase methods better names --- streamdeck/actions.py | 8 -------- streamdeck/models/events/__init__.py | 2 +- streamdeck/models/events/adapter.py | 4 +++- streamdeck/models/events/base.py | 20 +++++++++----------- tests/plugin_manager/test_plugin_manager.py | 2 +- 5 files changed, 14 insertions(+), 22 deletions(-) diff --git a/streamdeck/actions.py b/streamdeck/actions.py index 6161848..b49cd20 100644 --- a/streamdeck/actions.py +++ b/streamdeck/actions.py @@ -42,10 +42,6 @@ def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[TEvent_c Raises: KeyError: If the provided event name is not available. """ - # if event_name not in DEFAULT_EVENT_NAMES: - # msg = f"Provided event name for action handler does not exist: {event_name}" - # raise KeyError(msg) - 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)) @@ -66,10 +62,6 @@ def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHand Raises: KeyError: If the provided event name is not available. """ - # if event_name not in DEFAULT_EVENT_NAMES: - # msg = f"Provided event name for pulling handlers from action does not exist: {event_name}" - # raise KeyError(msg) - if event_name not in self._events: return diff --git a/streamdeck/models/events/__init__.py b/streamdeck/models/events/__init__.py index 75f86ef..be53204 100644 --- a/streamdeck/models/events/__init__.py +++ b/streamdeck/models/events/__init__.py @@ -60,7 +60,7 @@ def _get_default_event_names() -> set[str]: default_event_names: set[str] = set() for event_model in DEFAULT_EVENT_MODELS: - default_event_names.update(event_model.get_model_event_name()) + default_event_names.update(event_model.get_model_event_names()) return default_event_names diff --git a/streamdeck/models/events/adapter.py b/streamdeck/models/events/adapter.py index 347fa15..bf775b2 100644 --- a/streamdeck/models/events/adapter.py +++ b/streamdeck/models/events/adapter.py @@ -29,7 +29,9 @@ def __init__(self) -> None: def add_model(self, model: type[EventBase]) -> None: """Add a model to the adapter, and add the event name of the model to the set of registered event names.""" self._models.append(model) - self._event_names.update(model.get_model_event_name()) + # Models can have multiple event names defined in the Literal args of the event field, + # so `get_model_event_names()` returns a tuple of all event names, even if there is only one. + self._event_names.update(model.get_model_event_names()) def event_name_exists(self, event_name: str) -> bool: """Check if an event name has been registered with the adapter.""" diff --git a/streamdeck/models/events/base.py b/streamdeck/models/events/base.py index b003a76..99ad54d 100644 --- a/streamdeck/models/events/base.py +++ b/streamdeck/models/events/base.py @@ -1,12 +1,12 @@ from __future__ import annotations from abc import ABC -from typing import Any, Literal, get_args, get_type_hints +from typing import Any, Literal, get_args, get_origin, get_type_hints from pydantic import BaseModel, ConfigDict from typing_extensions import ( # noqa: UP035 LiteralString, - TypeIs, + TypeGuard, override, ) @@ -33,12 +33,12 @@ def model_dump_json(self, **kwargs: Any) -> str: return super().model_dump_json(**kwargs) -def is_literal_str_type(value: object | None) -> TypeIs[LiteralString]: +def is_literal_str_type(value: object | None) -> TypeGuard[LiteralString]: """Check if a type is a Literal type with string values.""" if value is None: return False - event_field_base_type = getattr(value, "__origin__", None) + event_field_base_type = get_origin(value) if event_field_base_type is not Literal: return False @@ -50,8 +50,6 @@ def is_literal_str_type(value: object | None) -> TypeIs[LiteralString]: class EventBase(ConfiguredBaseModel, ABC): """Base class for event models that represent Stream Deck Plugin SDK events.""" - # Configure to use the docstrings of the fields as the field descriptions. - model_config = ConfigDict(use_attribute_docstrings=True, serialize_by_alias=True) event: str """Name of the event used to identify what occurred. @@ -63,25 +61,25 @@ def __init_subclass__(cls, **kwargs: Any) -> None: """Validate that the event field is a Literal[str] type.""" super().__init_subclass__(**kwargs) - model_event_type = cls.get_event_type_annotations() + model_event_type = cls.__event_type__() if not is_literal_str_type(model_event_type): msg = f"The event field annotation must be a Literal[str] type. Given type: {model_event_type}" raise TypeError(msg) @classmethod - def get_event_type_annotations(cls) -> type[object]: + def __event_type__(cls) -> type[object]: """Get the type annotations of the subclass model's event field.""" return get_type_hints(cls)["event"] @classmethod - def get_model_event_name(cls) -> tuple[str, ...]: + def get_model_event_names(cls) -> tuple[str, ...]: """Get the value of the subclass model's event field Literal annotation.""" - model_event_type = cls.get_event_type_annotations() + model_event_type = cls.__event_type__() # Ensure that the event field annotation is a Literal type. if not is_literal_str_type(model_event_type): - msg = "The `event` field annotation of an Event model must be a Literal[str] type." + msg = f"The event field annotation of an Event model must be a Literal[str] type. Given type: {model_event_type}" raise TypeError(msg) return get_args(model_event_type) diff --git a/tests/plugin_manager/test_plugin_manager.py b/tests/plugin_manager/test_plugin_manager.py index 4900482..b471539 100644 --- a/tests/plugin_manager/test_plugin_manager.py +++ b/tests/plugin_manager/test_plugin_manager.py @@ -54,7 +54,7 @@ def test_plugin_manager_register_action(plugin_manager: PluginManager) -> None: def test_plugin_manager_register_event_listener(plugin_manager: PluginManager) -> None: """Test that an event listener can be registered in the PluginManager.""" - mock_event_model_class = Mock(get_model_event_name=lambda: ["fake_event_name"]) + mock_event_model_class = Mock(get_model_event_names=lambda: ["fake_event_name"]) listener = Mock(event_models=[mock_event_model_class]) assert len(plugin_manager._event_listener_manager.listeners_lookup_by_thread) == 0 From 3363d2d11be3b9bdd39647b2b8acddf668511857 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Wed, 9 Apr 2025 18:30:32 -0600 Subject: [PATCH 2/5] Refactor event field in EventBase to be generic, and update concrete base classes to handle this --- streamdeck/models/events/application.py | 6 ++--- streamdeck/models/events/base.py | 23 +++++++++++++++---- streamdeck/models/events/deep_link.py | 3 +-- streamdeck/models/events/devices.py | 6 ++--- streamdeck/models/events/dials.py | 9 +++----- streamdeck/models/events/keys.py | 6 ++--- .../models/events/property_inspector.py | 13 +++++------ streamdeck/models/events/settings.py | 6 ++--- streamdeck/models/events/system.py | 3 +-- streamdeck/models/events/title_parameters.py | 3 +-- streamdeck/models/events/touch_tap.py | 3 +-- streamdeck/models/events/visibility.py | 6 ++--- 12 files changed, 41 insertions(+), 46 deletions(-) diff --git a/streamdeck/models/events/application.py b/streamdeck/models/events/application.py index 6e4a788..c0d9dcf 100644 --- a/streamdeck/models/events/application.py +++ b/streamdeck/models/events/application.py @@ -11,15 +11,13 @@ class ApplicationPayload(ConfiguredBaseModel): """Name of the application that triggered the event.""" -class ApplicationDidLaunch(EventBase): +class ApplicationDidLaunch(EventBase[Literal["applicationDidLaunch"]]): """Occurs when a monitored application is launched.""" - event: Literal["applicationDidLaunch"] # type: ignore[override] payload: ApplicationPayload """Payload containing the name of the application that triggered the event.""" -class ApplicationDidTerminate(EventBase): +class ApplicationDidTerminate(EventBase[Literal["applicationDidTerminate"]]): """Occurs when a monitored application terminates.""" - event: Literal["applicationDidTerminate"] # type: ignore[override] payload: ApplicationPayload """Payload containing the name of the application that triggered the event.""" diff --git a/streamdeck/models/events/base.py b/streamdeck/models/events/base.py index 99ad54d..265f772 100644 --- a/streamdeck/models/events/base.py +++ b/streamdeck/models/events/base.py @@ -1,12 +1,14 @@ from __future__ import annotations from abc import ABC -from typing import Any, Literal, get_args, get_origin, get_type_hints +from typing import Any, Generic, Literal, get_args, get_origin from pydantic import BaseModel, ConfigDict +from pydantic._internal._generics import get_origin as get_model_origin from typing_extensions import ( # noqa: UP035 LiteralString, TypeGuard, + TypeVar, override, ) @@ -34,10 +36,13 @@ def model_dump_json(self, **kwargs: Any) -> str: def is_literal_str_type(value: object | None) -> TypeGuard[LiteralString]: - """Check if a type is a Literal type with string values.""" + """Check if a type is a concrete Literal type with string args.""" if value is None: return False + if isinstance(value, TypeVar): + return False + event_field_base_type = get_origin(value) if event_field_base_type is not Literal: @@ -48,10 +53,13 @@ def is_literal_str_type(value: object | None) -> TypeGuard[LiteralString]: ## EventBase implementation model of the Stream Deck Plugin SDK events. -class EventBase(ConfiguredBaseModel, ABC): +EventName_co = TypeVar("EventName_co", bound=str, default=str, covariant=True) + + +class EventBase(ConfiguredBaseModel, ABC, Generic[EventName_co]): """Base class for event models that represent Stream Deck Plugin SDK events.""" - event: str + event: EventName_co """Name of the event used to identify what occurred. Subclass models must define this field as a Literal type with the event name string that the model represents. @@ -61,6 +69,11 @@ def __init_subclass__(cls, **kwargs: Any) -> None: """Validate that the event field is a Literal[str] type.""" super().__init_subclass__(**kwargs) + # This is a GenericAlias (likely used in the subclass definition, i.e. `class ConcreteEvent(EventBase[Literal["event_name"]]):`) which is technically a subclass. + # We can safely ignore this case, as we only want to validate the concrete subclass itself (`ConscreteEvent`). + if get_model_origin(cls) is None: + return + model_event_type = cls.__event_type__() if not is_literal_str_type(model_event_type): @@ -70,7 +83,7 @@ def __init_subclass__(cls, **kwargs: Any) -> None: @classmethod def __event_type__(cls) -> type[object]: """Get the type annotations of the subclass model's event field.""" - return get_type_hints(cls)["event"] + return cls.model_fields["event"].annotation # type: ignore[index] @classmethod def get_model_event_names(cls) -> tuple[str, ...]: diff --git a/streamdeck/models/events/deep_link.py b/streamdeck/models/events/deep_link.py index 6331ce5..ae018ec 100644 --- a/streamdeck/models/events/deep_link.py +++ b/streamdeck/models/events/deep_link.py @@ -9,13 +9,12 @@ class DeepLinkPayload(ConfiguredBaseModel): """The deep-link URL, with the prefix omitted.""" -class DidReceiveDeepLink(EventBase): +class DidReceiveDeepLink(EventBase[Literal["didReceiveDeepLink"]]): """Occurs when Stream Deck receives a deep-link message intended for the plugin. The message is re-routed to the plugin, and provided as part of the payload. One-way deep-link message can be routed to the plugin using the URL format: streamdeck://plugins/message//{MESSAGE}. """ - event: Literal["didReceiveDeepLink"] # type: ignore[override] payload: DeepLinkPayload """Payload containing information about the URL that triggered the event.""" diff --git a/streamdeck/models/events/devices.py b/streamdeck/models/events/devices.py index 50c1ac9..cbd3beb 100644 --- a/streamdeck/models/events/devices.py +++ b/streamdeck/models/events/devices.py @@ -78,13 +78,11 @@ def __repr__(self) -> str: return f"DeviceInfo(name={self.name}, type={self.type}, size={self.size})" -class DeviceDidConnect(EventBase, DeviceSpecificEventMixin): +class DeviceDidConnect(EventBase[Literal["deviceDidConnect"]], DeviceSpecificEventMixin): """Occurs when a Stream Deck device is connected.""" - event: Literal["deviceDidConnect"] # type: ignore[override] device_info: Annotated[DeviceInfo, Field(alias="deviceInfo")] """Information about the newly connected device.""" -class DeviceDidDisconnect(EventBase, DeviceSpecificEventMixin): +class DeviceDidDisconnect(EventBase[Literal["deviceDidDisconnect"]], DeviceSpecificEventMixin): """Occurs when a Stream Deck device is disconnected.""" - event: Literal["deviceDidDisconnect"] # type: ignore[override] diff --git a/streamdeck/models/events/dials.py b/streamdeck/models/events/dials.py index ba63ada..8cf5154 100644 --- a/streamdeck/models/events/dials.py +++ b/streamdeck/models/events/dials.py @@ -28,22 +28,19 @@ class DialRotatePayload(EncoderPayload): ## Event models for DialDown, DialRotate, and DialUp events -class DialDown(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): +class DialDown(EventBase[Literal["dialDown"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user presses a dial (Stream Deck +).""" - event: Literal["dialDown"] # type: ignore[override] payload: EncoderPayload """Contextualized information for this event.""" -class DialRotate(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): +class DialRotate(EventBase[Literal["dialRotate"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user rotates a dial (Stream Deck +).""" - event: Literal["dialRotate"] # type: ignore[override] payload: DialRotatePayload """Contextualized information for this event.""" -class DialUp(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): +class DialUp(EventBase[Literal["dialUp"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user releases a pressed dial (Stream Deck +).""" - event: Literal["dialUp"] # type: ignore[override] payload: EncoderPayload """Contextualized information for this event.""" diff --git a/streamdeck/models/events/keys.py b/streamdeck/models/events/keys.py index 6b74921..0d7f44b 100644 --- a/streamdeck/models/events/keys.py +++ b/streamdeck/models/events/keys.py @@ -48,15 +48,13 @@ class MultiActionKeyGesturePayload( ## Event models for KeyDown and KeyUp events -class KeyDown(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): +class KeyDown(EventBase[Literal["keyDown"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user presses a action down.""" - event: Literal["keyDown"] # type: ignore[override] payload: CardinalityDiscriminated[SingleActionKeyGesturePayload, MultiActionKeyGesturePayload] """Contextualized information for this event.""" -class KeyUp(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): +class KeyUp(EventBase[Literal["keyUp"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user releases a pressed action.""" - event: Literal["keyUp"] # type: ignore[override] payload: CardinalityDiscriminated[SingleActionKeyGesturePayload, MultiActionKeyGesturePayload] """Contextualized information for this event.""" diff --git a/streamdeck/models/events/property_inspector.py b/streamdeck/models/events/property_inspector.py index 8097fd4..5fa6de2 100644 --- a/streamdeck/models/events/property_inspector.py +++ b/streamdeck/models/events/property_inspector.py @@ -2,7 +2,9 @@ from typing import Literal -from streamdeck.models.events.base import EventBase +from streamdeck.models.events.base import ( + EventBase, +) from streamdeck.models.events.common import ( ContextualEventMixin, DeviceSpecificEventMixin, @@ -10,24 +12,21 @@ ) -class DidReceivePropertyInspectorMessage(EventBase, ContextualEventMixin): +class DidReceivePropertyInspectorMessage(EventBase[Literal["sendToPlugin"]], ContextualEventMixin): """Occurs when a message was received from the UI.""" - event: Literal["sendToPlugin"] # type: ignore[override] payload: PluginDefinedData """The data payload received from the UI.""" -class PropertyInspectorDidAppear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): +class PropertyInspectorDidAppear(EventBase[Literal["propertyInspectorDidAppear"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the property inspector associated with the action becomes visible. I.e. the user selected an action in the Stream Deck application. """ - event: Literal["propertyInspectorDidAppear"] # type: ignore[override] -class PropertyInspectorDidDisappear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): +class PropertyInspectorDidDisappear(EventBase[Literal["propertyInspectorDidDisappear"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the property inspector associated with the action becomes invisible. I.e. the user unselected the action in the Stream Deck application. """ - event: Literal["propertyInspectorDidDisappear"] # type: ignore[override] diff --git a/streamdeck/models/events/settings.py b/streamdeck/models/events/settings.py index c9dab53..ec183f8 100644 --- a/streamdeck/models/events/settings.py +++ b/streamdeck/models/events/settings.py @@ -40,9 +40,8 @@ class MultiActionSettingsPayload( """ -class DidReceiveSettings(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): +class DidReceiveSettings(EventBase[Literal["didReceiveSettings"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the settings associated with an action instance are requested, or when the the settings were updated by the property inspector.""" - event: Literal["didReceiveSettings"] # type: ignore[override] payload: CardinalityDiscriminated[SingleActionSettingsPayload, MultiActionSettingsPayload] """Contextualized information for this event.""" @@ -55,8 +54,7 @@ class GlobalSettingsPayload(ConfiguredBaseModel): """The global settings received from the Stream Deck.""" -class DidReceiveGlobalSettings(EventBase): +class DidReceiveGlobalSettings(EventBase[Literal["didReceiveGlobalSettings"]]): """Occurs when the plugin receives the global settings from the Stream Deck.""" - event: Literal["didReceiveGlobalSettings"] # type: ignore[override] payload: GlobalSettingsPayload """Additional information about the event that occurred.""" diff --git a/streamdeck/models/events/system.py b/streamdeck/models/events/system.py index 2c6be51..040377d 100644 --- a/streamdeck/models/events/system.py +++ b/streamdeck/models/events/system.py @@ -5,6 +5,5 @@ from streamdeck.models.events.base import EventBase -class SystemDidWakeUp(EventBase): +class SystemDidWakeUp(EventBase[Literal["systemDidWakeUp"]]): """Occurs when the computer wakes up.""" - event: Literal["systemDidWakeUp"] # type: ignore[override] diff --git a/streamdeck/models/events/title_parameters.py b/streamdeck/models/events/title_parameters.py index 386d089..97f9e0f 100644 --- a/streamdeck/models/events/title_parameters.py +++ b/streamdeck/models/events/title_parameters.py @@ -47,9 +47,8 @@ class TitleParametersDidChangePayload( """Defines aesthetic properties that determine how the title should be rendered.""" -class TitleParametersDidChange(EventBase, DeviceSpecificEventMixin): +class TitleParametersDidChange(EventBase[Literal["titleParametersDidChange"]], DeviceSpecificEventMixin): """Occurs when the user updates an action's title settings in the Stream Deck application.""" - event: Literal["titleParametersDidChange"] # type: ignore[override] context: str """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" payload: TitleParametersDidChangePayload diff --git a/streamdeck/models/events/touch_tap.py b/streamdeck/models/events/touch_tap.py index 4226a1c..1b1402d 100644 --- a/streamdeck/models/events/touch_tap.py +++ b/streamdeck/models/events/touch_tap.py @@ -22,8 +22,7 @@ class TouchTapPayload(BasePayload[EncoderControllerType], CoordinatesPayloadMixi """Coordinates of where the touchscreen tap occurred, relative to the canvas of the action.""" -class TouchTap(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): +class TouchTap(EventBase[Literal["touchTap"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user taps the touchscreen (Stream Deck +).""" - event: Literal["touchTap"] # type: ignore[override] payload: TouchTapPayload """Contextualized information for this event.""" diff --git a/streamdeck/models/events/visibility.py b/streamdeck/models/events/visibility.py index cf35c34..4f50533 100644 --- a/streamdeck/models/events/visibility.py +++ b/streamdeck/models/events/visibility.py @@ -35,20 +35,18 @@ class MultiActionVisibilityPayload( ## Event models for WillAppear and WillDisappear events -class WillAppear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): +class WillAppear(EventBase[Literal["willAppear"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when an action appears on the Stream Deck due to the user navigating to another page, profile, folder, etc. This also occurs during startup if the action is on the "front page". An action refers to all types of actions, e.g. keys, dials, touchscreens, pedals, etc. """ - event: Literal["willAppear"] # type: ignore[override] payload: CardinalityDiscriminated[SingleActionVisibilityPayload, MultiActionVisibilityPayload] -class WillDisappear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): +class WillDisappear(EventBase[Literal["willDisappear"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when an action disappears from the Stream Deck due to the user navigating to another page, profile, folder, etc. An action refers to all types of actions, e.g. keys, dials, touchscreens, pedals, etc. """ - event: Literal["willDisappear"] # type: ignore[override] payload: CardinalityDiscriminated[SingleActionVisibilityPayload, MultiActionVisibilityPayload] From 0a0170d24776171e3ba7a8b3e1624d41498e1891 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Thu, 10 Apr 2025 18:19:30 -0600 Subject: [PATCH 3/5] Replace str TypeVar with a Pydantic-friendly LiteralGenericAlias for the generic EventBase --- streamdeck/models/events/base.py | 37 ++++++++++++++++++++++++++------ 1 file changed, 30 insertions(+), 7 deletions(-) diff --git a/streamdeck/models/events/base.py b/streamdeck/models/events/base.py index 265f772..6285940 100644 --- a/streamdeck/models/events/base.py +++ b/streamdeck/models/events/base.py @@ -1,12 +1,14 @@ from __future__ import annotations from abc import ABC -from typing import Any, Generic, Literal, get_args, get_origin +from typing import Annotated, Any, Generic, Literal, get_args, get_origin -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel, ConfigDict, GetPydanticSchema from pydantic._internal._generics import get_origin as get_model_origin +from pydantic_core import core_schema from typing_extensions import ( # noqa: UP035 LiteralString, + TypeAlias, TypeGuard, TypeVar, override, @@ -35,6 +37,30 @@ def model_dump_json(self, **kwargs: Any) -> str: return super().model_dump_json(**kwargs) +# We do this to get the typing module's _LiteralGenericAlias type, which is not formally exported. +LiteralGenericAlias: TypeAlias = type(Literal["whatever"]) # type: ignore[valid-type] # noqa: UP040 +"""A generic alias for a Literal type.""" + +_pydantic_str_schema = core_schema.str_schema() + +PydanticLiteralStrGenericAlias: TypeAlias = Annotated[ # type: ignore[valid-type] # noqa: UP040 + LiteralGenericAlias, + GetPydanticSchema(lambda _ts, handler: handler(_pydantic_str_schema)) +] +"""A Pydantic-compatible generic alias for a Literal type. + +Pydantic will treat a field of this type as a string schema, while static type checkers +will still treat it as a LiteralGenericAlias type. + +Even if a subclass of EventBase uses a Literal with multiple string values, +an event message will only ever have one of those values in the event field, +and so we don't need to handle this with a more complex Pydantic schema. +""" + +LiteralEventName_co = TypeVar("LiteralEventName_co", bound=PydanticLiteralStrGenericAlias, default=PydanticLiteralStrGenericAlias, covariant=True) +"""Type variable for a Literal type with string args.""" + + def is_literal_str_type(value: object | None) -> TypeGuard[LiteralString]: """Check if a type is a concrete Literal type with string args.""" if value is None: @@ -53,13 +79,10 @@ def is_literal_str_type(value: object | None) -> TypeGuard[LiteralString]: ## EventBase implementation model of the Stream Deck Plugin SDK events. -EventName_co = TypeVar("EventName_co", bound=str, default=str, covariant=True) - - -class EventBase(ConfiguredBaseModel, ABC, Generic[EventName_co]): +class EventBase(ConfiguredBaseModel, ABC, Generic[LiteralEventName_co]): """Base class for event models that represent Stream Deck Plugin SDK events.""" - event: EventName_co + event: LiteralEventName_co """Name of the event used to identify what occurred. Subclass models must define this field as a Literal type with the event name string that the model represents. From da37951932c4d8c55a1c2c5628972c2a16bfc573 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Thu, 10 Apr 2025 19:08:13 -0600 Subject: [PATCH 4/5] Update references to EventBase with type argument of LiteralStrGenericAlias --- streamdeck/actions.py | 6 +++--- streamdeck/event_listener.py | 4 +++- streamdeck/manager.py | 5 +++-- streamdeck/models/events/__init__.py | 4 +++- streamdeck/models/events/adapter.py | 12 ++++++------ streamdeck/models/events/base.py | 19 +++++++++++-------- streamdeck/types.py | 9 ++++++--- .../test_action_event_handler_filtering.py | 17 +++++++++-------- tests/actions/test_action_registry.py | 7 ++++--- tests/actions/test_actions.py | 13 +++++++------ 10 files changed, 55 insertions(+), 41 deletions(-) diff --git a/streamdeck/actions.py b/streamdeck/actions.py index b49cd20..215bfbe 100644 --- a/streamdeck/actions.py +++ b/streamdeck/actions.py @@ -6,7 +6,7 @@ from logging import getLogger from typing import TYPE_CHECKING, cast -from streamdeck.types import BaseEventHandlerFunc +from streamdeck.types import BaseEventHandlerFunc, LiteralStrGenericAlias if TYPE_CHECKING: @@ -50,7 +50,7 @@ def _wrapper(func: EventHandlerFunc[TEvent_contra]) -> EventHandlerFunc[TEvent_c 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[EventBase[LiteralStrGenericAlias]], None, None]: """Get all event handlers for a specific event. Args: @@ -112,7 +112,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[EventBase[LiteralStrGenericAlias]], None, None]: """Get all event handlers for a specific event from all registered actions. Args: diff --git a/streamdeck/event_listener.py b/streamdeck/event_listener.py index 2e1da2e..d0db2ef 100644 --- a/streamdeck/event_listener.py +++ b/streamdeck/event_listener.py @@ -6,6 +6,8 @@ from queue import Queue from typing import TYPE_CHECKING +from streamdeck.types import LiteralStrGenericAlias + if TYPE_CHECKING: from collections.abc import Generator @@ -119,7 +121,7 @@ class EventListener(ABC): Event listeners are classes that listen for events and simply yield them as they come. The EventListenerManager will handle the threading and pushing the events yielded into a shared queue. """ - event_models: ClassVar[list[type[EventBase]]] + event_models: ClassVar[list[type[EventBase[LiteralStrGenericAlias]]]] """A list of event models that the listener can yield. Read in by the PluginManager to model the incoming event data off of. The plugin-developer must define this list in their subclass. diff --git a/streamdeck/manager.py b/streamdeck/manager.py index 7746a8d..df3877c 100644 --- a/streamdeck/manager.py +++ b/streamdeck/manager.py @@ -14,6 +14,7 @@ from streamdeck.types import ( EventHandlerBasicFunc, EventHandlerFunc, + LiteralStrGenericAlias, TEvent_contra, is_bindable_handler, ) @@ -120,7 +121,7 @@ def _inject_command_sender(self, handler: EventHandlerFunc[TEvent_contra], comma return handler - def _stream_event_data(self) -> Generator[EventBase, None, None]: + def _stream_event_data(self) -> Generator[EventBase[LiteralStrGenericAlias], None, None]: """Stream event data from the event listeners. Validate and model the incoming event data before yielding it. @@ -130,7 +131,7 @@ def _stream_event_data(self) -> Generator[EventBase, None, None]: """ for message in self._event_listener_manager.event_stream(): try: - data: EventBase = self._event_adapter.validate_json(message) + data: EventBase[LiteralStrGenericAlias] = self._event_adapter.validate_json(message) except ValidationError: logger.exception("Error modeling event data.") continue diff --git a/streamdeck/models/events/__init__.py b/streamdeck/models/events/__init__.py index be53204..81b3b6d 100644 --- a/streamdeck/models/events/__init__.py +++ b/streamdeck/models/events/__init__.py @@ -31,8 +31,10 @@ if TYPE_CHECKING: from typing import Final + from streamdeck.types import LiteralStrGenericAlias -DEFAULT_EVENT_MODELS: Final[list[type[EventBase]]] = [ + +DEFAULT_EVENT_MODELS: Final[list[type[EventBase[LiteralStrGenericAlias]]]] = [ ApplicationDidLaunch, ApplicationDidTerminate, DeviceDidConnect, diff --git a/streamdeck/models/events/adapter.py b/streamdeck/models/events/adapter.py index bf775b2..23519ee 100644 --- a/streamdeck/models/events/adapter.py +++ b/streamdeck/models/events/adapter.py @@ -5,18 +5,18 @@ from pydantic import Field, TypeAdapter from streamdeck.models.events import DEFAULT_EVENT_MODELS +from streamdeck.types import LiteralStrGenericAlias if TYPE_CHECKING: from streamdeck.models.events.base import EventBase - class EventAdapter: """TypeAdapter-encompassing class for handling and extending available event models.""" def __init__(self) -> None: - self._models: list[type[EventBase]] = [] - self._type_adapter: TypeAdapter[EventBase] | None = None + self._models: list[type[EventBase[LiteralStrGenericAlias]]] = [] + self._type_adapter: TypeAdapter[EventBase[LiteralStrGenericAlias]] | None = None self._event_names: set[str] = set() """A set of all event names that have been registered with the adapter. @@ -26,7 +26,7 @@ def __init__(self) -> None: for model in DEFAULT_EVENT_MODELS: self.add_model(model) - def add_model(self, model: type[EventBase]) -> None: + def add_model(self, model: type[EventBase[LiteralStrGenericAlias]]) -> None: """Add a model to the adapter, and add the event name of the model to the set of registered event names.""" self._models.append(model) # Models can have multiple event names defined in the Literal args of the event field, @@ -38,7 +38,7 @@ def event_name_exists(self, event_name: str) -> bool: return event_name in self._event_names @property - def type_adapter(self) -> TypeAdapter[EventBase]: + def type_adapter(self) -> TypeAdapter[EventBase[LiteralStrGenericAlias]]: """Get the TypeAdapter instance for the event models.""" if self._type_adapter is None: self._type_adapter = TypeAdapter( @@ -50,7 +50,7 @@ def type_adapter(self) -> TypeAdapter[EventBase]: return self._type_adapter - def validate_json(self, data: str | bytes) -> EventBase: + def validate_json(self, data: str | bytes) -> EventBase[LiteralStrGenericAlias]: """Validate a JSON string or bytes object as an event model.""" return self.type_adapter.validate_json(data) diff --git a/streamdeck/models/events/base.py b/streamdeck/models/events/base.py index 6285940..48a714c 100644 --- a/streamdeck/models/events/base.py +++ b/streamdeck/models/events/base.py @@ -4,7 +4,7 @@ from typing import Annotated, Any, Generic, Literal, get_args, get_origin from pydantic import BaseModel, ConfigDict, GetPydanticSchema -from pydantic._internal._generics import get_origin as get_model_origin +from pydantic._internal._generics import get_origin as get_model_origin # type: ignore[import] from pydantic_core import core_schema from typing_extensions import ( # noqa: UP035 LiteralString, @@ -38,19 +38,22 @@ def model_dump_json(self, **kwargs: Any) -> str: # We do this to get the typing module's _LiteralGenericAlias type, which is not formally exported. -LiteralGenericAlias: TypeAlias = type(Literal["whatever"]) # type: ignore[valid-type] # noqa: UP040 -"""A generic alias for a Literal type.""" +_LiteralGenericAlias: TypeAlias = type(Literal["whatever"]) # type: ignore[valid-type] # noqa: UP040 +"""A generic alias for a Literal type used for internal mechanisms of this module. + +This is opposed to the streamdeck.types.LiteralStrGenericAlias which is used for typing. +""" _pydantic_str_schema = core_schema.str_schema() PydanticLiteralStrGenericAlias: TypeAlias = Annotated[ # type: ignore[valid-type] # noqa: UP040 - LiteralGenericAlias, + _LiteralGenericAlias, GetPydanticSchema(lambda _ts, handler: handler(_pydantic_str_schema)) ] """A Pydantic-compatible generic alias for a Literal type. Pydantic will treat a field of this type as a string schema, while static type checkers -will still treat it as a LiteralGenericAlias type. +will still treat it as a _LiteralGenericAlias type. Even if a subclass of EventBase uses a Literal with multiple string values, an event message will only ever have one of those values in the event field, @@ -61,7 +64,7 @@ def model_dump_json(self, **kwargs: Any) -> str: """Type variable for a Literal type with string args.""" -def is_literal_str_type(value: object | None) -> TypeGuard[LiteralString]: +def is_literal_str_generic_alias_type(value: object | None) -> TypeGuard[LiteralString]: """Check if a type is a concrete Literal type with string args.""" if value is None: return False @@ -99,7 +102,7 @@ def __init_subclass__(cls, **kwargs: Any) -> None: model_event_type = cls.__event_type__() - if not is_literal_str_type(model_event_type): + if not is_literal_str_generic_alias_type(model_event_type): msg = f"The event field annotation must be a Literal[str] type. Given type: {model_event_type}" raise TypeError(msg) @@ -114,7 +117,7 @@ def get_model_event_names(cls) -> tuple[str, ...]: model_event_type = cls.__event_type__() # Ensure that the event field annotation is a Literal type. - if not is_literal_str_type(model_event_type): + if not is_literal_str_generic_alias_type(model_event_type): msg = f"The event field annotation of an Event model must be a Literal[str] type. Given type: {model_event_type}" raise TypeError(msg) diff --git a/streamdeck/types.py b/streamdeck/types.py index a2d3994..0d82f39 100644 --- a/streamdeck/types.py +++ b/streamdeck/types.py @@ -3,6 +3,8 @@ import inspect from typing import TYPE_CHECKING, Protocol, TypeVar, Union +from typing_extensions import LiteralString # noqa: UP035 + from streamdeck.models.events import DEFAULT_EVENT_NAMES, EventBase @@ -20,9 +22,10 @@ """ -def is_valid_event_name(event_name: str) -> TypeIs[EventNameStr]: - """Check if the event name is one of the available event names.""" - return event_name in DEFAULT_EVENT_NAMES +# This type alias is used to handle static type checking accurately while still conveying that +# a value is expected to be a Literal with string type args. +LiteralStrGenericAlias: TypeAlias = LiteralString # noqa: UP040 +"""Type alias for a generic literal string type.""" ### Event Handler Type Definitions ### diff --git a/tests/actions/test_action_event_handler_filtering.py b/tests/actions/test_action_event_handler_filtering.py index d55915d..591f5ee 100644 --- a/tests/actions/test_action_event_handler_filtering.py +++ b/tests/actions/test_action_event_handler_filtering.py @@ -6,6 +6,7 @@ import pytest from streamdeck.actions import Action, ActionRegistry, GlobalAction from streamdeck.models.events.common import ContextualEventMixin +from streamdeck.types import LiteralStrGenericAlias from tests.test_utils.fake_event_factories import ( ApplicationDidLaunchEventFactory, @@ -35,14 +36,14 @@ def dummy_handler(event: events.EventBase) -> None: DeviceDidConnectFactory, ApplicationDidLaunchEventFactory ]) -def fake_event_data(request: pytest.FixtureRequest) -> events.EventBase: - event_factory = cast("ModelFactory[events.EventBase]", request.param) +def fake_event_data(request: pytest.FixtureRequest) -> events.EventBase[LiteralStrGenericAlias]: + event_factory = cast("ModelFactory[events.EventBase[LiteralStrGenericAlias]]", request.param) return event_factory.build() def test_global_action_gets_triggered_by_event( mock_event_handler: Mock, - fake_event_data: events.EventBase, + fake_event_data: events.EventBase[LiteralStrGenericAlias], ) -> None: """Test that a global action's event handlers are triggered by an event. @@ -62,7 +63,7 @@ def test_global_action_gets_triggered_by_event( def test_action_gets_triggered_by_event( mock_event_handler: Mock, - fake_event_data: events.EventBase, + fake_event_data: events.EventBase[LiteralStrGenericAlias], ) -> None: """Test that an action's event handlers are triggered by an event. @@ -89,7 +90,7 @@ def test_action_gets_triggered_by_event( def test_global_action_registry_get_action_handlers_filtering( mock_event_handler: Mock, - fake_event_data: events.EventBase, + fake_event_data: events.EventBase[LiteralStrGenericAlias], ) -> None: # Extract the action UUID from the fake event data, or use a default value action_uuid: str | None = fake_event_data.action if isinstance(fake_event_data, ContextualEventMixin) else None @@ -116,7 +117,7 @@ def test_global_action_registry_get_action_handlers_filtering( def test_action_registry_get_action_handlers_filtering( mock_event_handler: Mock, - fake_event_data: events.EventBase, + fake_event_data: events.EventBase[LiteralStrGenericAlias], ) -> None: # Extract the action UUID from the fake event data, or use a default value action_uuid: str | None = fake_event_data.action if isinstance(fake_event_data, ContextualEventMixin) else None @@ -150,12 +151,12 @@ def test_multiple_actions_filtering() -> None: action_event_handler_called = False @global_action.on("applicationDidLaunch") - def _global_app_did_launch_action_handler(event: events.EventBase) -> None: + def _global_app_did_launch_action_handler(event: events.EventBase[LiteralStrGenericAlias]) -> None: nonlocal global_action_event_handler_called global_action_event_handler_called = True @action.on("keyDown") - def _action_key_down_event_handler(event: events.EventBase) -> None: + def _action_key_down_event_handler(event: events.EventBase[LiteralStrGenericAlias]) -> None: nonlocal action_event_handler_called action_event_handler_called = True diff --git a/tests/actions/test_action_registry.py b/tests/actions/test_action_registry.py index 606d820..d6522de 100644 --- a/tests/actions/test_action_registry.py +++ b/tests/actions/test_action_registry.py @@ -4,6 +4,7 @@ import pytest from streamdeck.actions import Action, ActionRegistry +from streamdeck.types import LiteralStrGenericAlias from tests.test_utils.fake_event_factories import ( DialDownEventFactory, @@ -46,7 +47,7 @@ def test_get_action_handlers_with_handlers() -> None: action = Action("my-fake-action-uuid") @action.on("dialDown") - def dial_down_handler(event: events.EventBase) -> None: + def dial_down_handler(event_data: events.EventBase[LiteralStrGenericAlias]) -> None: pass registry.register(action) @@ -66,11 +67,11 @@ def test_get_action_handlers_multiple_actions() -> None: action2 = Action("fake-action-uuid-2") @action1.on("keyUp") - def key_up_handler1(event) -> None: + def key_up_handler1(event_data: events.EventBase[LiteralStrGenericAlias]) -> None: pass @action2.on("keyUp") - def key_up_handler2(event) -> None: + def key_up_handler2(event_data: events.EventBase[LiteralStrGenericAlias]) -> None: pass registry.register(action1) diff --git a/tests/actions/test_actions.py b/tests/actions/test_actions.py index 85660a4..cb986d2 100644 --- a/tests/actions/test_actions.py +++ b/tests/actions/test_actions.py @@ -5,13 +5,14 @@ import pytest from streamdeck.actions import Action, ActionBase, GlobalAction from streamdeck.models.events import DEFAULT_EVENT_NAMES +from streamdeck.types import LiteralStrGenericAlias if TYPE_CHECKING: from streamdeck.models.events import EventBase -@pytest.fixture(params=[[Action, ("test.uuid.for.action",)], [GlobalAction, []]]) +@pytest.fixture(params=[[Action, ("test.uuid.for.action",)], [GlobalAction, ()]]) def action(request: pytest.FixtureRequest) -> ActionBase: """Fixture for initializing the Action and GlobalAction classes to parameterize the tests. @@ -25,7 +26,7 @@ def action(request: pytest.FixtureRequest) -> ActionBase: def test_action_register_event_handler(action: ActionBase, event_name: str) -> None: """Test that an event handler can be registered for each valid event name.""" @action.on(event_name) - def handler(event: EventBase) -> None: + def handler(event_data: EventBase[LiteralStrGenericAlias]) -> None: pass # Ensure the handler is registered for the correct event name @@ -37,10 +38,10 @@ def test_action_get_event_handlers(action: ActionBase) -> None: """Test that the correct event handlers are retrieved for each event name.""" # Each iteration will add to the action's event handlers, thus we're checking that # even with multiple event names, the handlers are correctly retrieved. - for i, event_name in enumerate(DEFAULT_EVENT_NAMES): + for _, event_name in enumerate(DEFAULT_EVENT_NAMES): # Register a handler for the given event name @action.on(event_name) - def handler(event: EventBase) -> None: + def handler(event_data: EventBase[LiteralStrGenericAlias]) -> None: pass # Retrieve the handlers using the generator @@ -73,11 +74,11 @@ def test_action_get_event_handlers_invalid_event_name(action: ActionBase) -> Non def test_action_register_multiple_handlers_for_event(action: ActionBase) -> None: """Test that multiple handlers can be registered for the same event on the same action.""" @action.on("keyDown") - def handler_one(event: EventBase) -> None: + def handler_one(event_data: EventBase[LiteralStrGenericAlias]) -> None: pass @action.on("keyDown") - def handler_two(event: EventBase) -> None: + def handler_two(event_data: EventBase[LiteralStrGenericAlias]) -> None: pass handlers = list(action.get_event_handlers("keyDown")) From 5bb632704a369e02139670e0b18bc7961002390a Mon Sep 17 00:00:00 2001 From: strohganoff Date: Thu, 10 Apr 2025 19:34:15 -0600 Subject: [PATCH 5/5] Move LiteralStrGenericAlias to base event module, and make it Pydantic-compatible. Update import references --- streamdeck/actions.py | 5 ++-- streamdeck/event_listener.py | 3 +-- streamdeck/manager.py | 2 +- streamdeck/models/events/__init__.py | 2 +- streamdeck/models/events/adapter.py | 3 +-- streamdeck/models/events/base.py | 26 +++++++++++++++---- streamdeck/types.py | 10 +------ .../test_action_event_handler_filtering.py | 2 +- tests/actions/test_action_registry.py | 2 +- tests/actions/test_actions.py | 2 +- tests/event_listener/test_event_listener.py | 15 ++++++----- tests/plugin_manager/conftest.py | 3 ++- .../test_command_sender_binding.py | 11 ++++---- tests/plugin_manager/test_plugin_manager.py | 3 ++- 14 files changed, 49 insertions(+), 40 deletions(-) diff --git a/streamdeck/actions.py b/streamdeck/actions.py index 215bfbe..f69682c 100644 --- a/streamdeck/actions.py +++ b/streamdeck/actions.py @@ -6,14 +6,13 @@ from logging import getLogger from typing import TYPE_CHECKING, cast -from streamdeck.types import BaseEventHandlerFunc, LiteralStrGenericAlias - if TYPE_CHECKING: from collections.abc import Callable, Generator from streamdeck.models.events import EventBase - from streamdeck.types import EventHandlerFunc, EventNameStr, TEvent_contra + from streamdeck.models.events.base import LiteralStrGenericAlias + from streamdeck.types import BaseEventHandlerFunc, EventHandlerFunc, EventNameStr, TEvent_contra logger = getLogger("streamdeck.actions") diff --git a/streamdeck/event_listener.py b/streamdeck/event_listener.py index d0db2ef..0aba535 100644 --- a/streamdeck/event_listener.py +++ b/streamdeck/event_listener.py @@ -6,8 +6,6 @@ from queue import Queue from typing import TYPE_CHECKING -from streamdeck.types import LiteralStrGenericAlias - if TYPE_CHECKING: from collections.abc import Generator @@ -16,6 +14,7 @@ from typing_extensions import TypeIs from streamdeck.models.events import EventBase + from streamdeck.models.events.base import LiteralStrGenericAlias diff --git a/streamdeck/manager.py b/streamdeck/manager.py index df3877c..4eae3f8 100644 --- a/streamdeck/manager.py +++ b/streamdeck/manager.py @@ -14,7 +14,6 @@ from streamdeck.types import ( EventHandlerBasicFunc, EventHandlerFunc, - LiteralStrGenericAlias, TEvent_contra, is_bindable_handler, ) @@ -27,6 +26,7 @@ from typing import Any, Literal from streamdeck.models.events import EventBase + from streamdeck.models.events.base import LiteralStrGenericAlias # TODO: Fix this up to push to a log in the apropos directory and filename. diff --git a/streamdeck/models/events/__init__.py b/streamdeck/models/events/__init__.py index 81b3b6d..d5a7fad 100644 --- a/streamdeck/models/events/__init__.py +++ b/streamdeck/models/events/__init__.py @@ -31,7 +31,7 @@ if TYPE_CHECKING: from typing import Final - from streamdeck.types import LiteralStrGenericAlias + from streamdeck.models.events.base import LiteralStrGenericAlias DEFAULT_EVENT_MODELS: Final[list[type[EventBase[LiteralStrGenericAlias]]]] = [ diff --git a/streamdeck/models/events/adapter.py b/streamdeck/models/events/adapter.py index 23519ee..6509f19 100644 --- a/streamdeck/models/events/adapter.py +++ b/streamdeck/models/events/adapter.py @@ -5,11 +5,10 @@ from pydantic import Field, TypeAdapter from streamdeck.models.events import DEFAULT_EVENT_MODELS -from streamdeck.types import LiteralStrGenericAlias if TYPE_CHECKING: - from streamdeck.models.events.base import EventBase + from streamdeck.models.events.base import EventBase, LiteralStrGenericAlias class EventAdapter: diff --git a/streamdeck/models/events/base.py b/streamdeck/models/events/base.py index 48a714c..78f291f 100644 --- a/streamdeck/models/events/base.py +++ b/streamdeck/models/events/base.py @@ -38,17 +38,22 @@ def model_dump_json(self, **kwargs: Any) -> str: # We do this to get the typing module's _LiteralGenericAlias type, which is not formally exported. -_LiteralGenericAlias: TypeAlias = type(Literal["whatever"]) # type: ignore[valid-type] # noqa: UP040 +_LiteralStrGenericAlias: TypeAlias = type(Literal["whatever"]) # type: ignore[valid-type] # noqa: UP040 """A generic alias for a Literal type used for internal mechanisms of this module. -This is opposed to the streamdeck.types.LiteralStrGenericAlias which is used for typing. +This is opposed to LiteralStrGenericAlias which is used for typing. """ + +# Set this variable here to call the function just once. _pydantic_str_schema = core_schema.str_schema() +GetPydanticStrSchema = GetPydanticSchema(lambda _ts, handler: handler(_pydantic_str_schema)) +"""A function that returns a Pydantic schema for a string type.""" + PydanticLiteralStrGenericAlias: TypeAlias = Annotated[ # type: ignore[valid-type] # noqa: UP040 - _LiteralGenericAlias, - GetPydanticSchema(lambda _ts, handler: handler(_pydantic_str_schema)) + _LiteralStrGenericAlias, + GetPydanticStrSchema, ] """A Pydantic-compatible generic alias for a Literal type. @@ -60,11 +65,22 @@ def model_dump_json(self, **kwargs: Any) -> str: and so we don't need to handle this with a more complex Pydantic schema. """ + +# This type alias is used to handle static type checking accurately while still conveying that +# a value is expected to be a Literal with string type args. +LiteralStrGenericAlias: TypeAlias = Annotated[ # noqa: UP040 + LiteralString, + GetPydanticStrSchema, +] +"""Type alias for a generic literal string type that is compatible with Pydantic.""" + + +# covariant=True is used to allow subclasses of EventBase to be used in place of the base class. LiteralEventName_co = TypeVar("LiteralEventName_co", bound=PydanticLiteralStrGenericAlias, default=PydanticLiteralStrGenericAlias, covariant=True) """Type variable for a Literal type with string args.""" -def is_literal_str_generic_alias_type(value: object | None) -> TypeGuard[LiteralString]: +def is_literal_str_generic_alias_type(value: object | None) -> TypeGuard[LiteralStrGenericAlias]: """Check if a type is a concrete Literal type with string args.""" if value is None: return False diff --git a/streamdeck/types.py b/streamdeck/types.py index 0d82f39..d8e37dc 100644 --- a/streamdeck/types.py +++ b/streamdeck/types.py @@ -3,9 +3,7 @@ import inspect from typing import TYPE_CHECKING, Protocol, TypeVar, Union -from typing_extensions import LiteralString # noqa: UP035 - -from streamdeck.models.events import DEFAULT_EVENT_NAMES, EventBase +from streamdeck.models.events import EventBase if TYPE_CHECKING: @@ -22,12 +20,6 @@ """ -# This type alias is used to handle static type checking accurately while still conveying that -# a value is expected to be a Literal with string type args. -LiteralStrGenericAlias: TypeAlias = LiteralString # noqa: UP040 -"""Type alias for a generic literal string type.""" - - ### Event Handler Type Definitions ### ## Protocols for event handler functions that act on subtypes of EventBase instances in a Generic way. diff --git a/tests/actions/test_action_event_handler_filtering.py b/tests/actions/test_action_event_handler_filtering.py index 591f5ee..9941a9b 100644 --- a/tests/actions/test_action_event_handler_filtering.py +++ b/tests/actions/test_action_event_handler_filtering.py @@ -6,7 +6,6 @@ import pytest from streamdeck.actions import Action, ActionRegistry, GlobalAction from streamdeck.models.events.common import ContextualEventMixin -from streamdeck.types import LiteralStrGenericAlias from tests.test_utils.fake_event_factories import ( ApplicationDidLaunchEventFactory, @@ -20,6 +19,7 @@ from polyfactory.factories.pydantic_factory import ModelFactory from streamdeck.models import events + from streamdeck.models.events.base import LiteralStrGenericAlias diff --git a/tests/actions/test_action_registry.py b/tests/actions/test_action_registry.py index d6522de..abf9a3a 100644 --- a/tests/actions/test_action_registry.py +++ b/tests/actions/test_action_registry.py @@ -4,7 +4,6 @@ import pytest from streamdeck.actions import Action, ActionRegistry -from streamdeck.types import LiteralStrGenericAlias from tests.test_utils.fake_event_factories import ( DialDownEventFactory, @@ -15,6 +14,7 @@ if TYPE_CHECKING: from streamdeck.models import events + from streamdeck.models.events.base import LiteralStrGenericAlias def test_register_action() -> None: diff --git a/tests/actions/test_actions.py b/tests/actions/test_actions.py index cb986d2..5ce8a20 100644 --- a/tests/actions/test_actions.py +++ b/tests/actions/test_actions.py @@ -5,11 +5,11 @@ import pytest from streamdeck.actions import Action, ActionBase, GlobalAction from streamdeck.models.events import DEFAULT_EVENT_NAMES -from streamdeck.types import LiteralStrGenericAlias if TYPE_CHECKING: from streamdeck.models.events import EventBase + from streamdeck.models.events.base import LiteralStrGenericAlias @pytest.fixture(params=[[Action, ("test.uuid.for.action",)], [GlobalAction, ()]]) diff --git a/tests/event_listener/test_event_listener.py b/tests/event_listener/test_event_listener.py index 0ba9782..d2bf1b9 100644 --- a/tests/event_listener/test_event_listener.py +++ b/tests/event_listener/test_event_listener.py @@ -7,11 +7,12 @@ import pytest from streamdeck.event_listener import EventListener, EventListenerManager from streamdeck.models.events import ApplicationDidLaunch, EventBase +from streamdeck.models.events.base import LiteralStrGenericAlias class MockEventListener(EventListener): """Mock implementation of EventListener for testing.""" - event_models: ClassVar[list[type[EventBase]]] = [ApplicationDidLaunch] + event_models: ClassVar[list[type[EventBase[LiteralStrGenericAlias]]]] = [ApplicationDidLaunch] def __init__(self): self._running = True @@ -33,7 +34,7 @@ def stop(self) -> None: class SlowMockEventListener(EventListener): """Mock implementation of EventListener that yields events with a delay.""" - event_models: ClassVar[list[type[EventBase]]] = [ApplicationDidLaunch] + event_models: ClassVar[list[type[EventBase[LiteralStrGenericAlias]]]] = [ApplicationDidLaunch] def __init__(self, delay: float = 0.1): self._running = True @@ -53,7 +54,7 @@ def stop(self) -> None: class ExceptionEventListener(EventListener): """Mock implementation of EventListener that raises an exception.""" - event_models: ClassVar[list[type[EventBase]]] = [ApplicationDidLaunch] + event_models: ClassVar[list[type[EventBase[LiteralStrGenericAlias]]]] = [ApplicationDidLaunch] def listen(self) -> Generator[str, None, None]: self._running = True @@ -91,9 +92,9 @@ def test_event_stream_basic(): manager.add_listener(listener) # Collect the first few events - events = [] + events : list[EventNameStr] = [] for event in manager.event_stream(): - events.append(event) + events.append(event) # type: ignore[arg-type] if len(events) >= 3: # We expect 3 events from MockEventListener manager.stop() break @@ -112,9 +113,9 @@ def test_event_stream_multiple_listeners(): manager.add_listener(listener2) # Collect all events - events = [] + events: list[EventNameStr] = [] for event in manager.event_stream(): - events.append(event) + events.append(event) # type: ignore[arg-type] if len(events) >= 6: # We expect 6 events total (3 from each listener) manager.stop() break diff --git a/tests/plugin_manager/conftest.py b/tests/plugin_manager/conftest.py index 02a4194..87c02aa 100644 --- a/tests/plugin_manager/conftest.py +++ b/tests/plugin_manager/conftest.py @@ -14,6 +14,7 @@ if TYPE_CHECKING: import pytest_mock from streamdeck.models import events + from streamdeck.models.events.base import LiteralStrGenericAlias @pytest.fixture @@ -103,7 +104,7 @@ def mock_event_listener_manager_with_fake_events(patch_event_listener_manager: M """ print("MOCK EVENT LISTENER MANAGER") # Create a list of fake event messages, and convert them to json strings to be passed back by the client.listen() method. - fake_event_messages: list[events.EventBase] = [ + fake_event_messages: list[events.EventBase[LiteralStrGenericAlias]] = [ KeyDownEventFactory.build(action="my-fake-action-uuid"), ] diff --git a/tests/plugin_manager/test_command_sender_binding.py b/tests/plugin_manager/test_command_sender_binding.py index 70f2090..b137c63 100644 --- a/tests/plugin_manager/test_command_sender_binding.py +++ b/tests/plugin_manager/test_command_sender_binding.py @@ -14,6 +14,7 @@ from streamdeck.command_sender import StreamDeckCommandSender from streamdeck.manager import PluginManager from streamdeck.models import events + from streamdeck.models.events.base import LiteralStrGenericAlias from streamdeck.types import ( EventHandlerBasicFunc, EventHandlerFunc, @@ -31,12 +32,12 @@ def create_event_handler(include_command_sender_param: bool = False) -> EventHan Callable[[events.EventBase], None] | Callable[[events.EventBase, StreamDeckCommandSender], None]: A dummy event handler function. """ if not include_command_sender_param: - def dummy_handler_without_cmd_sender(event_data: events.EventBase) -> None: + def dummy_handler_without_cmd_sender(event_data: events.EventBase[LiteralStrGenericAlias]) -> None: """Dummy event handler function that matches the EventHandlerFunc TypeAlias without `command_sender` param.""" return dummy_handler_without_cmd_sender - def dummy_handler_with_cmd_sender(event_data: events.EventBase, command_sender: StreamDeckCommandSender) -> None: + def dummy_handler_with_cmd_sender(event_data: events.EventBase[LiteralStrGenericAlias], command_sender: StreamDeckCommandSender) -> None: """Dummy event handler function that matches the EventHandlerFunc TypeAlias with `command_sender` param.""" return dummy_handler_with_cmd_sender @@ -45,7 +46,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[events.EventBase[LiteralStrGenericAlias]] = create_event_handler(include_command_sender_param) return create_autospec(dummy_handler, spec_set=True) @@ -57,7 +58,7 @@ def test_inject_command_sender_func( ) -> None: """Test that the command_sender is injected into the handler.""" mock_command_sender = Mock() - result_handler: EventHandlerBasicFunc[events.EventBase] = plugin_manager._inject_command_sender(mock_event_handler, mock_command_sender) + result_handler: EventHandlerBasicFunc[events.EventBase[LiteralStrGenericAlias]] = plugin_manager._inject_command_sender(mock_event_handler, mock_command_sender) resulting_handler_params = inspect.signature(result_handler).parameters @@ -78,7 +79,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]], + mock_event_listener_manager_with_fake_events: tuple[Mock, list[events.EventBase[LiteralStrGenericAlias]]], plugin_manager: PluginManager, mock_command_sender: Mock, ) -> None: diff --git a/tests/plugin_manager/test_plugin_manager.py b/tests/plugin_manager/test_plugin_manager.py index b471539..d152fa0 100644 --- a/tests/plugin_manager/test_plugin_manager.py +++ b/tests/plugin_manager/test_plugin_manager.py @@ -10,6 +10,7 @@ DEFAULT_EVENT_NAMES, EventBase, ) +from streamdeck.models.events.base import LiteralStrGenericAlias @pytest.fixture @@ -115,7 +116,7 @@ def test_plugin_manager_adds_websocket_event_listener( @pytest.mark.integration @pytest.mark.usefixtures("patch_websocket_client") def test_plugin_manager_process_event( - mock_event_listener_manager_with_fake_events: tuple[Mock, list[EventBase]], + mock_event_listener_manager_with_fake_events: tuple[Mock, list[EventBase[LiteralStrGenericAlias]]], _spy_action_registry_get_action_handlers: None, # This fixture must come after mock_event_listener_manager_with_fake_events to ensure monkeypatching occurs. _spy_event_adapter_validate_json: None, # This fixture must come after mock_event_listener_manager_with_fake_events to ensure monkeypatching occurs. plugin_manager: PluginManager, # This fixture must come after patch_event_listener_manager and spy-fixtures to ensure things are mocked and spied correctly.