Make event models generic over literal names

README.md

@@ -323,7 +323,7 @@ The SDK allows you to create custom event listeners and events by extending the
 
 To create a custom event listener:
 
-1. Create new event model that inherits from `EventBase["eventName"]`.
+1. Create new event model that inherits from `EventBase[Literal["eventName"]]`.
 2. Create a new class that inherits from `EventListener`.
     a. Implement the required `listen` and `stop` methods. The `listen` method should yield results as a json string that matches the new event model.
     b. List the new event classes in the `event_models` class variable of the new `EventListener` class.
@@ -338,7 +338,7 @@ from streamdeck.event_listener import EventListener
 from streamdeck.models.events import EventBase
 
 
-class MyCustomEvent(EventBase["somethingHappened"]):
+class MyCustomEvent(EventBase[Literal["somethingHappened"]]):
     # The 'event' field's type annotation is internally set as Literal["somethingHappened"]
     # Define additional data attributes here
     result: str

streamdeck/models/configs.py

@@ -112,5 +112,10 @@ def event_models(self) -> list[type[EventBase]]:
         return [
             obj
             for obj in self._event_listener_objects
-            if isinstance(obj, type) and obj is not EventBase and issubclass(obj, EventBase)
+            if (
+                isinstance(obj, type)
+                and obj is not EventBase  # Exclude the base class itself.
+                and issubclass(obj, EventBase)
+                and getattr(obj, "__pydantic_generic_metadata__", {}).get("origin") is not EventBase
+            )
         ]

streamdeck/models/events/adapter.py

@@ -1,13 +1,15 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Annotated, Union
+from typing import TYPE_CHECKING, Annotated, Union, cast
 
 from pydantic import Field, TypeAdapter
 
 from streamdeck.models.events import DEFAULT_EVENT_MODELS
 
 
 if TYPE_CHECKING:
+    from typing_extensions import TypeForm
+
     from streamdeck.models.events.base import EventBase
     from streamdeck.types import EventNameStr
 
@@ -41,11 +43,15 @@ def event_name_exists(self, event_name: str) -> bool:
     def type_adapter(self) -> TypeAdapter[EventBase]:
         """Get the TypeAdapter instance for the event models."""
         if self._type_adapter is None:
+            # Mypy cannot infer the type form of this union because its models are collected at runtime.
             self._type_adapter = TypeAdapter(
-                Annotated[
-                    Union[tuple(self._models)],  # noqa: UP007
-                    Field(discriminator="event")
-                ]
+                cast(
+                    "TypeForm[EventBase]",
+                    Annotated[
+                        Union[tuple(self._models)],  # noqa: UP007
+                        Field(discriminator="event")
+                    ],
+                )
             )
 
         return self._type_adapter

streamdeck/models/events/application.py

@@ -1,21 +1,26 @@
 from __future__ import annotations
 
+from typing import Literal
+
 from streamdeck.models.events.base import ConfiguredBaseModel, EventBase
 
 
 class ApplicationPayload(ConfiguredBaseModel):
     """Payload containing the name of the application that triggered the event."""
+
     application: str
     """Name of the application that triggered the event."""
 
 
-class ApplicationDidLaunch(EventBase["applicationDidLaunch"]):
+class ApplicationDidLaunch(EventBase[Literal["applicationDidLaunch"]]):
     """Occurs when a monitored application is launched."""
+
     payload: ApplicationPayload
     """Payload containing the name of the application that triggered the event."""
 
 
-class ApplicationDidTerminate(EventBase["applicationDidTerminate"]):
+class ApplicationDidTerminate(EventBase[Literal["applicationDidTerminate"]]):
     """Occurs when a monitored application terminates."""
+
     payload: ApplicationPayload
     """Payload containing the name of the application that triggered the event."""

streamdeck/models/events/base.py

@@ -1,20 +1,30 @@
+"""Shared Pydantic base classes for incoming event models."""
+
 from __future__ import annotations
 
 from abc import ABC
-from typing import TYPE_CHECKING, Literal
-from weakref import WeakValueDictionary
+from typing import TYPE_CHECKING, Annotated, Any, Generic, Literal, get_args, get_origin
 
-from pydantic import BaseModel, ConfigDict, create_model
-from typing_extensions import LiteralString, TypedDict, override  # noqa: UP035
+from pydantic import BaseModel, ConfigDict, GetPydanticSchema
+from typing_extensions import LiteralString, TypeVar, override
 
 
 if TYPE_CHECKING:
-    from typing import Any, ClassVar
+    from typing import ClassVar, TypeAlias
+
+
+StringLiteralType: TypeAlias = Annotated[
+    LiteralString,
+    GetPydanticSchema(lambda _source_type, handler: handler(str)),
+]
+"""A string-valued Literal type, with a Pydantic schema fallback to str."""
 
+EventNameT = TypeVar("EventNameT", bound=StringLiteralType, default=Any)
 
 
 class ConfiguredBaseModel(BaseModel, ABC):
     """Base class for models that share the same configuration."""
+
     # Configure to use the docstrings of the fields as the field descriptions,
     # and to serialize the fields by their aliases.
     model_config = ConfigDict(use_attribute_docstrings=True, serialize_by_alias=True)
@@ -35,180 +45,74 @@ def model_dump_json(self, **kwargs: Any) -> str:
         return super().model_dump_json(**kwargs)
 
 
-class EventMetadataDict(TypedDict):
-    """Metadata for specialized EventBase submodels.
+class EventBase(ConfiguredBaseModel, Generic[EventNameT]):
+    """Base class for all event models.
 
-    Similar to the __pydantic_generic_metadata__ attribute, but for use in the EventBase class——which isn't actually generic.
+    EventBase itself and its generic specializations should not be instantiated.
+    Define concrete event models by specializing with a string-valued Literal,
+    for example `class KeyDown(EventBase[Literal["keyDown"]]): ...`.
     """
-    origin: type[EventBase]
-    """Origin class of the specialized EventBase submodel."""
-    args: tuple[str, ...]
-    """Event names for the specialized EventBase submodel."""
-
-
-if TYPE_CHECKING:
-    # Because we can't override a BaseModel's metaclass __getitem__ method without angering Pydantic during runtime,
-    # we define this stub here to satisfy static type checkers that introspect the metaclass method annotations
-    # to determine expected types in the class subscriptions.
-
-    from collections.abc import Callable
-
-    from pydantic._internal._model_construction import ModelMetaclass
-
-    from streamdeck.types import EventNameStr
-
-    class EventMeta(ModelMetaclass):
-        """Metaclass for EventBase stub to satisfy static type checkers."""
-        @classmethod
-        def __getitem__(cls, event_names: LiteralString | tuple[LiteralString, ...]) -> type[EventBase]: ...
 
-    class EventBase(BaseModel, metaclass=EventMeta):
-        """Base class for all event models.
+    event: EventNameT
+    """Name of the event used to identify what occurred."""
 
-        EventBase itself should not be instantiated, nor should it be subclassed directly.
-        Instead, use a subscripted subclass of EventBase, e.g. `EventBase["eventName"]`, to subclass from.
+    __event_names__: ClassVar[tuple[str, ...]] = ()
+    """The event names declared by the generic EventBase specialization."""
 
-        Examples:
-        ```
-        class KeyDown(EventBase["keyDown"]):
-            # 'event' field's type annotation is internally set here as `Literal["keyDown"]`
-            ...
+    @classmethod
+    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:  # noqa: ANN401
+        """Validate EventBase generic arguments and require a specialization on subclasses."""
+        super().__pydantic_init_subclass__(**kwargs)
 
-        class TestEvent(EventBase["test", "testing"]):
-            # 'event' field's type annotation is internally set here as `Literal["test", "testing"]`
-            ...
+        if cls is EventBase:
+            return
 
-        ```
-        """
-        event: EventNameStr
-        """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.
-        """
-        __event_metadata__: ClassVar[EventMetadataDict]
-        """Metadata for specialized EventBase submodels."""
-        __event_type__: ClassVar[Callable[[], type[object] | None]]
-        """Return the event type for the event model."""
-
-        @classmethod
-        def get_model_event_names(cls) -> tuple[EventNameStr, ...]:
-            """Return the event names for the event model."""
-            ...
-
-else:
-    class EventBase(ConfiguredBaseModel, ABC):
-        """Base class for all event models.
-
-        EventBase itself should not be instantiated, nor should it be subclassed directly.
-        Instead, use a subscripted subclass of EventBase, e.g. `EventBase["eventName"]`, to subclass from.
-
-        Examples:
-        ```
-        class KeyDown(EventBase["keyDown"]):
-            # 'event' field's type annotation is internally set here as `Literal["keyDown"]`
-            ...
-
-        class TestEvent(EventBase["test", "testing"]):
-            # 'event' field's type annotation is internally set here as `Literal["test", "testing"]`
-            ...
-
-        ```
-        """
-        # A weak reference dictionary to store subscripted subclasses of EventBase. Weak references are used to minimize memory usage.
-        _cached_specialized_submodels: ClassVar[WeakValueDictionary[str, type[EventBase]]] = WeakValueDictionary()
-        __event_metadata__: ClassVar[EventMetadataDict]
-        """Metadata for specialized EventBase submodels."""
-
-        event: str
-        """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.
-        """
-
-        def __new__(cls, *args: Any, **kwargs: Any) -> EventBase:
-            """Create a new instance of the event model.
-
-            This method is called when the model is instantiated.
-            It validates that the event name is a string and not empty.
-            """
-            if cls is EventBase:
-                raise TypeError("Can't instantiate abstract class EventBase directly.")
-            if EventBase in cls.__bases__:
-                raise TypeError(f"Can't instantiate abstract subscripted EventBase class {cls.__name__} directly.")
-            return super().__new__(cls)
-
-        @override
-        @classmethod
-        def __class_getitem__(cls: type[EventBase], event_names: LiteralString | tuple[LiteralString, ...]) -> type[EventBase]: # type: ignore[override]
-            """Create a subclass alias of EventBase with the given event names.
-
-            This method is called when the class is subscripted.
-            If the event names are not already registered, a new subclass is created.
-            Validations are performed to ensure that the event names are strings and not empty, and that the subscripted class is not a subclass of EventBase.
-            """
-            if cls is not EventBase: # type: ignore[misc]
-                raise TypeError(f"Subclasses of EventBase are not subscriptable — '{cls.__name__}'")  # noqa: TRY003, EM102
-
-            # Whether event_names is a string or tuple of strings, validate that it/they is/are a string
-            if not isinstance(event_names, tuple):
-                event_names = (event_names,)
-
-            # Validate that all event names are strings
-            if any(not isinstance(name, str) for name in event_names): # type: ignore[misc]
-                raise TypeError(f"Event names must be strings, not {type(event_names).__name__}; args: {event_names}")  # noqa: TRY003, EM102
-
-            subtype_name = f"{cls.__name__}[{event_names}]"
-
-            return cls.__new_subscripted_base__(subtype_name, event_names)
-
-        @classmethod
-        def __new_subscripted_base__(cls: type[EventBase], new_name: str, event_name_args: tuple[str, ...]) -> type[EventBase]:
-            """Dynamically create a new Singleton subclass of EventBase with the given event names for the event field.
-
-            Only create a new subscripted subclass if it doesn't already exist in the _cached_specialized_submodels dictionary, otherwise return the existing subclass.
-            The new subclasses created here will be ignored in the __init_subclass__ method.
-            """
-            if new_name not in cls._cached_specialized_submodels:
-                # Make sure not to pass in a value `_cached_specialized_submodels` in the create_model() call, in order to avoid shadowing the class variable.
-                cls._cached_specialized_submodels[new_name] = create_model(
-                    new_name,
-                    __base__=cls,
-                    __event_metadata__=(EventMetadataDict, {"origin": cls, "args": event_name_args}),
-                    event=(Literal[event_name_args], ...),
-                    __cls_kwargs__={"_is_specialized_base": True},  # This gets passed to __init_subclass__ as a kwarg to indicate that this is a specialized (subscripted) subclass of EventBase.
-                )
-
-            return cls._cached_specialized_submodels[new_name]
-
-        @classmethod
-        def __init_subclass__(cls, _is_specialized_base: bool = False) -> None:
-            """Validate a child class of EventBase (not a subscripted base subclass) is subclassing from a subscripted EventBase.
-
-            Args:
-                _is_specialized_base: Whether this is a specialized submodel of EventBase (i.e., a subscripted subclass).
-                    This should only be True for the subscripted subclasses created in __class_getitem__.
-            """
-            if _is_specialized_base:
-                # This is a subscripted subclass of EventBase, so we don't need to do anything.
-                return
-
-            if EventBase in cls.__bases__:
-                # Normally, only subscripted subclasses of EventBase (filtered out above) will have (un-subscripted) EventBase in their __bases__.
-                # If we get here, it means that this is a subclass of EventBase that is not subscripted, which is not allowed.
-                msg = f"Child classes of EventBase cannot subclass from non-subscripted EventBase. '{cls.__name__}' must subclass from a subscripted EventBase."
+        metadata = cls.__pydantic_generic_metadata__
+        if metadata.get("origin") is EventBase:
+            args = metadata["args"]
+            if len(args) != 1 or get_origin(args[0]) is not Literal:
+                msg = "EventBase must be specialized with Literal[str, ...]"
                 raise TypeError(msg)
 
-        def __str__(self) -> str:
-            """Return a string representation of the event model in the format 'EventName(event=event_name, field_name=field_value, etc.)'."""
-            field_key_value_strs: list[str] = [f"{field_name}={getattr(self, field_name)}" for field_name in type(self).model_fields]
-            return f"{self.__class__.__name__}({', '.join(field_key_value_strs)})"
-
-        @classmethod
-        def __event_type__(cls) -> type[object] | None:
-            """Return the event type for the event model."""
-            return cls.model_fields["event"].annotation
+            names = get_args(args[0])
+            if not names or any(type(name) is not str for name in names):
+                msg = "EventBase Literal arguments must be strings"
+                raise TypeError(msg)
 
-        @classmethod
-        def get_model_event_names(cls) -> tuple[str, ...]:
-            """Return the event names for the event model."""
-            return cls.__event_metadata__["args"]
+            cls.__event_names__ = names
+            return
+
+        if not any(
+            getattr(base, "__pydantic_generic_metadata__", {}).get("origin") is EventBase
+            for base in cls.__mro__[1:]
+        ):
+            msg = (
+                "Child classes of EventBase must subclass from a specialized EventBase. "
+                f"'{cls.__name__}' must use EventBase[Literal[...]]."
+            )
+            raise TypeError(msg)
+
+    def __new__(cls, *args: Any, **kwargs: Any) -> EventBase[Any]:  # noqa: ANN401, ARG004
+        """Prevent instantiation of EventBase and its generic specializations."""
+        metadata = cls.__pydantic_generic_metadata__
+        if cls is EventBase or metadata.get("origin") is EventBase:
+            msg = f"Can't instantiate abstract event model {cls.__name__} directly."
+            raise TypeError(msg)
+        return super().__new__(cls)
+
+    def __str__(self) -> str:
+        """Return a readable representation of the event model and its fields."""
+        field_key_value_strs = [
+            f"{field_name}={getattr(self, field_name)}" for field_name in type(self).model_fields
+        ]
+        return f"{self.__class__.__name__}({', '.join(field_key_value_strs)})"
+
+    @classmethod
+    def __event_type__(cls) -> Any:  # noqa: ANN401
+        """Return the Pydantic field annotation for this event model."""
+        return cls.model_fields["event"].annotation
+
+    @classmethod
+    def get_model_event_names(cls) -> tuple[str, ...]:
+        """Return the event names declared by this model's EventBase specialization."""
+        return cls.__event_names__

streamdeck/models/events/deep_link.py

@@ -1,18 +1,22 @@
+from typing import Literal
+
 from streamdeck.models.events.base import ConfiguredBaseModel, EventBase
 
 
 class DeepLinkPayload(ConfiguredBaseModel):
     """Payload containing information about the URL that triggered the event."""
+
     url: str
     """The deep-link URL, with the prefix omitted."""
 
 
-class DidReceiveDeepLink(EventBase["didReceiveDeepLink"]):
+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}.
     """
+
     payload: DeepLinkPayload
     """Payload containing information about the URL that triggered the event."""