Refactor/generic event name field

streamdeck/actions.py

@@ -6,14 +6,13 @@
 from logging import getLogger
 from typing import TYPE_CHECKING, cast
 
-from streamdeck.types import BaseEventHandlerFunc
-
 
 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")
@@ -42,10 +41,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))
@@ -54,7 +49,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:
@@ -66,10 +61,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
 
@@ -120,7 +111,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:

streamdeck/event_listener.py

@@ -14,6 +14,7 @@
     from typing_extensions import TypeIs
 
     from streamdeck.models.events import EventBase
+    from streamdeck.models.events.base import LiteralStrGenericAlias
 
 
 
@@ -119,7 +120,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.

streamdeck/manager.py

@@ -26,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.
@@ -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

streamdeck/models/events/__init__.py

@@ -31,8 +31,10 @@
 if TYPE_CHECKING:
     from typing import Final
 
+    from streamdeck.models.events.base import LiteralStrGenericAlias
 
-DEFAULT_EVENT_MODELS: Final[list[type[EventBase]]] = [
+
+DEFAULT_EVENT_MODELS: Final[list[type[EventBase[LiteralStrGenericAlias]]]] = [
     ApplicationDidLaunch,
     ApplicationDidTerminate,
     DeviceDidConnect,
@@ -60,7 +62,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
 

streamdeck/models/events/adapter.py

@@ -8,15 +8,14 @@
 
 
 if TYPE_CHECKING:
-    from streamdeck.models.events.base import EventBase
-
+    from streamdeck.models.events.base import EventBase, LiteralStrGenericAlias
 
 
 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,17 +25,19 @@ 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)
-        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."""
         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(
@@ -48,7 +49,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)
 

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."""

streamdeck/models/events/base.py

@@ -1,12 +1,16 @@
 from __future__ import annotations
 
 from abc import ABC
-from typing import Any, Literal, get_args, get_type_hints
+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  # type: ignore[import]
+from pydantic_core import core_schema
 from typing_extensions import (  # noqa: UP035
     LiteralString,
-    TypeIs,
+    TypeAlias,
+    TypeGuard,
+    TypeVar,
     override,
 )
 
@@ -33,12 +37,58 @@ def model_dump_json(self, **kwargs: Any) -> str:
         return super().model_dump_json(**kwargs)
 
 
-def is_literal_str_type(value: object | None) -> TypeIs[LiteralString]:
-    """Check if a type is a Literal type with string values."""
+# We do this to get the typing module's _LiteralGenericAlias type, which is not formally exported.
+_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 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
+    _LiteralStrGenericAlias,
+    GetPydanticStrSchema,
+]
+"""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.
+"""
+
+
+# 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[LiteralStrGenericAlias]:
+    """Check if a type is a concrete Literal type with string args."""
     if value is None:
         return False
 
-    event_field_base_type = getattr(value, "__origin__", None)
+    if isinstance(value, TypeVar):
+        return False
+
+    event_field_base_type = get_origin(value)
 
     if event_field_base_type is not Literal:
         return False
@@ -48,12 +98,10 @@ def is_literal_str_type(value: object | None) -> TypeIs[LiteralString]:
 
 ## EventBase implementation model of the Stream Deck Plugin SDK events.
 
-class EventBase(ConfiguredBaseModel, ABC):
+class EventBase(ConfiguredBaseModel, ABC, Generic[LiteralEventName_co]):
     """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
+    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.
@@ -63,25 +111,30 @@ 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()
+        # 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):
+        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)
 
     @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"]
+        return cls.model_fields["event"].annotation  # type: ignore[index]
 
     @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."
+        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)
 
         return get_args(model_event_type)

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/<PLUGIN_UUID>/{MESSAGE}.
     """
-    event: Literal["didReceiveDeepLink"]  # type: ignore[override]
     payload: DeepLinkPayload
     """Payload containing information about the URL that triggered the event."""

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]

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."""