Skip to content

Commit 060497e

Browse files
authored
Merge pull request #14 from strohganoff/refactor/generic-event-name-field
Refactor/generic event name field
2 parents 4582bb2 + 5bb6327 commit 060497e

25 files changed

+156
-123
lines changed

streamdeck/actions.py

+4-13
Original file line numberDiff line numberDiff line change
@@ -6,14 +6,13 @@
66
from logging import getLogger
77
from typing import TYPE_CHECKING, cast
88

9-
from streamdeck.types import BaseEventHandlerFunc
10-
119

1210
if TYPE_CHECKING:
1311
from collections.abc import Callable, Generator
1412

1513
from streamdeck.models.events import EventBase
16-
from streamdeck.types import EventHandlerFunc, EventNameStr, TEvent_contra
14+
from streamdeck.models.events.base import LiteralStrGenericAlias
15+
from streamdeck.types import BaseEventHandlerFunc, EventHandlerFunc, EventNameStr, TEvent_contra
1716

1817

1918
logger = getLogger("streamdeck.actions")
@@ -42,10 +41,6 @@ def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[TEvent_c
4241
Raises:
4342
KeyError: If the provided event name is not available.
4443
"""
45-
# if event_name not in DEFAULT_EVENT_NAMES:
46-
# msg = f"Provided event name for action handler does not exist: {event_name}"
47-
# raise KeyError(msg)
48-
4944
def _wrapper(func: EventHandlerFunc[TEvent_contra]) -> EventHandlerFunc[TEvent_contra]:
5045
# Cast to BaseEventHandlerFunc so that the storage type is consistent.
5146
self._events[event_name].add(cast("BaseEventHandlerFunc", func))
@@ -54,7 +49,7 @@ def _wrapper(func: EventHandlerFunc[TEvent_contra]) -> EventHandlerFunc[TEvent_c
5449

5550
return _wrapper
5651

57-
def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHandlerFunc[EventBase], None, None]:
52+
def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHandlerFunc[EventBase[LiteralStrGenericAlias]], None, None]:
5853
"""Get all event handlers for a specific event.
5954
6055
Args:
@@ -66,10 +61,6 @@ def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHand
6661
Raises:
6762
KeyError: If the provided event name is not available.
6863
"""
69-
# if event_name not in DEFAULT_EVENT_NAMES:
70-
# msg = f"Provided event name for pulling handlers from action does not exist: {event_name}"
71-
# raise KeyError(msg)
72-
7364
if event_name not in self._events:
7465
return
7566

@@ -120,7 +111,7 @@ def register(self, action: ActionBase) -> None:
120111
"""
121112
self._plugin_actions.append(action)
122113

123-
def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: str | None = None) -> Generator[EventHandlerFunc[EventBase], None, None]:
114+
def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: str | None = None) -> Generator[EventHandlerFunc[EventBase[LiteralStrGenericAlias]], None, None]:
124115
"""Get all event handlers for a specific event from all registered actions.
125116
126117
Args:

streamdeck/event_listener.py

+2-1
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
from typing_extensions import TypeIs
1515

1616
from streamdeck.models.events import EventBase
17+
from streamdeck.models.events.base import LiteralStrGenericAlias
1718

1819

1920

@@ -119,7 +120,7 @@ class EventListener(ABC):
119120
Event listeners are classes that listen for events and simply yield them as they come.
120121
The EventListenerManager will handle the threading and pushing the events yielded into a shared queue.
121122
"""
122-
event_models: ClassVar[list[type[EventBase]]]
123+
event_models: ClassVar[list[type[EventBase[LiteralStrGenericAlias]]]]
123124
"""A list of event models that the listener can yield. Read in by the PluginManager to model the incoming event data off of.
124125
125126
The plugin-developer must define this list in their subclass.

streamdeck/manager.py

+3-2
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
from typing import Any, Literal
2727

2828
from streamdeck.models.events import EventBase
29+
from streamdeck.models.events.base import LiteralStrGenericAlias
2930

3031

3132
# 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
120121

121122
return handler
122123

123-
def _stream_event_data(self) -> Generator[EventBase, None, None]:
124+
def _stream_event_data(self) -> Generator[EventBase[LiteralStrGenericAlias], None, None]:
124125
"""Stream event data from the event listeners.
125126
126127
Validate and model the incoming event data before yielding it.
@@ -130,7 +131,7 @@ def _stream_event_data(self) -> Generator[EventBase, None, None]:
130131
"""
131132
for message in self._event_listener_manager.event_stream():
132133
try:
133-
data: EventBase = self._event_adapter.validate_json(message)
134+
data: EventBase[LiteralStrGenericAlias] = self._event_adapter.validate_json(message)
134135
except ValidationError:
135136
logger.exception("Error modeling event data.")
136137
continue

streamdeck/models/events/__init__.py

+4-2
Original file line numberDiff line numberDiff line change
@@ -31,8 +31,10 @@
3131
if TYPE_CHECKING:
3232
from typing import Final
3333

34+
from streamdeck.models.events.base import LiteralStrGenericAlias
3435

35-
DEFAULT_EVENT_MODELS: Final[list[type[EventBase]]] = [
36+
37+
DEFAULT_EVENT_MODELS: Final[list[type[EventBase[LiteralStrGenericAlias]]]] = [
3638
ApplicationDidLaunch,
3739
ApplicationDidTerminate,
3840
DeviceDidConnect,
@@ -60,7 +62,7 @@ def _get_default_event_names() -> set[str]:
6062
default_event_names: set[str] = set()
6163

6264
for event_model in DEFAULT_EVENT_MODELS:
63-
default_event_names.update(event_model.get_model_event_name())
65+
default_event_names.update(event_model.get_model_event_names())
6466

6567
return default_event_names
6668

streamdeck/models/events/adapter.py

+9-8
Original file line numberDiff line numberDiff line change
@@ -8,15 +8,14 @@
88

99

1010
if TYPE_CHECKING:
11-
from streamdeck.models.events.base import EventBase
12-
11+
from streamdeck.models.events.base import EventBase, LiteralStrGenericAlias
1312

1413

1514
class EventAdapter:
1615
"""TypeAdapter-encompassing class for handling and extending available event models."""
1716
def __init__(self) -> None:
18-
self._models: list[type[EventBase]] = []
19-
self._type_adapter: TypeAdapter[EventBase] | None = None
17+
self._models: list[type[EventBase[LiteralStrGenericAlias]]] = []
18+
self._type_adapter: TypeAdapter[EventBase[LiteralStrGenericAlias]] | None = None
2019

2120
self._event_names: set[str] = set()
2221
"""A set of all event names that have been registered with the adapter.
@@ -26,17 +25,19 @@ def __init__(self) -> None:
2625
for model in DEFAULT_EVENT_MODELS:
2726
self.add_model(model)
2827

29-
def add_model(self, model: type[EventBase]) -> None:
28+
def add_model(self, model: type[EventBase[LiteralStrGenericAlias]]) -> None:
3029
"""Add a model to the adapter, and add the event name of the model to the set of registered event names."""
3130
self._models.append(model)
32-
self._event_names.update(model.get_model_event_name())
31+
# Models can have multiple event names defined in the Literal args of the event field,
32+
# so `get_model_event_names()` returns a tuple of all event names, even if there is only one.
33+
self._event_names.update(model.get_model_event_names())
3334

3435
def event_name_exists(self, event_name: str) -> bool:
3536
"""Check if an event name has been registered with the adapter."""
3637
return event_name in self._event_names
3738

3839
@property
39-
def type_adapter(self) -> TypeAdapter[EventBase]:
40+
def type_adapter(self) -> TypeAdapter[EventBase[LiteralStrGenericAlias]]:
4041
"""Get the TypeAdapter instance for the event models."""
4142
if self._type_adapter is None:
4243
self._type_adapter = TypeAdapter(
@@ -48,7 +49,7 @@ def type_adapter(self) -> TypeAdapter[EventBase]:
4849

4950
return self._type_adapter
5051

51-
def validate_json(self, data: str | bytes) -> EventBase:
52+
def validate_json(self, data: str | bytes) -> EventBase[LiteralStrGenericAlias]:
5253
"""Validate a JSON string or bytes object as an event model."""
5354
return self.type_adapter.validate_json(data)
5455

streamdeck/models/events/application.py

+2-4
Original file line numberDiff line numberDiff line change
@@ -11,15 +11,13 @@ class ApplicationPayload(ConfiguredBaseModel):
1111
"""Name of the application that triggered the event."""
1212

1313

14-
class ApplicationDidLaunch(EventBase):
14+
class ApplicationDidLaunch(EventBase[Literal["applicationDidLaunch"]]):
1515
"""Occurs when a monitored application is launched."""
16-
event: Literal["applicationDidLaunch"] # type: ignore[override]
1716
payload: ApplicationPayload
1817
"""Payload containing the name of the application that triggered the event."""
1918

2019

21-
class ApplicationDidTerminate(EventBase):
20+
class ApplicationDidTerminate(EventBase[Literal["applicationDidTerminate"]]):
2221
"""Occurs when a monitored application terminates."""
23-
event: Literal["applicationDidTerminate"] # type: ignore[override]
2422
payload: ApplicationPayload
2523
"""Payload containing the name of the application that triggered the event."""

streamdeck/models/events/base.py

+71-18
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,16 @@
11
from __future__ import annotations
22

33
from abc import ABC
4-
from typing import Any, Literal, get_args, get_type_hints
4+
from typing import Annotated, Any, Generic, Literal, get_args, get_origin
55

6-
from pydantic import BaseModel, ConfigDict
6+
from pydantic import BaseModel, ConfigDict, GetPydanticSchema
7+
from pydantic._internal._generics import get_origin as get_model_origin # type: ignore[import]
8+
from pydantic_core import core_schema
79
from typing_extensions import ( # noqa: UP035
810
LiteralString,
9-
TypeIs,
11+
TypeAlias,
12+
TypeGuard,
13+
TypeVar,
1014
override,
1115
)
1216

@@ -33,12 +37,58 @@ def model_dump_json(self, **kwargs: Any) -> str:
3337
return super().model_dump_json(**kwargs)
3438

3539

36-
def is_literal_str_type(value: object | None) -> TypeIs[LiteralString]:
37-
"""Check if a type is a Literal type with string values."""
40+
# We do this to get the typing module's _LiteralGenericAlias type, which is not formally exported.
41+
_LiteralStrGenericAlias: TypeAlias = type(Literal["whatever"]) # type: ignore[valid-type] # noqa: UP040
42+
"""A generic alias for a Literal type used for internal mechanisms of this module.
43+
44+
This is opposed to LiteralStrGenericAlias which is used for typing.
45+
"""
46+
47+
48+
# Set this variable here to call the function just once.
49+
_pydantic_str_schema = core_schema.str_schema()
50+
51+
GetPydanticStrSchema = GetPydanticSchema(lambda _ts, handler: handler(_pydantic_str_schema))
52+
"""A function that returns a Pydantic schema for a string type."""
53+
54+
PydanticLiteralStrGenericAlias: TypeAlias = Annotated[ # type: ignore[valid-type] # noqa: UP040
55+
_LiteralStrGenericAlias,
56+
GetPydanticStrSchema,
57+
]
58+
"""A Pydantic-compatible generic alias for a Literal type.
59+
60+
Pydantic will treat a field of this type as a string schema, while static type checkers
61+
will still treat it as a _LiteralGenericAlias type.
62+
63+
Even if a subclass of EventBase uses a Literal with multiple string values,
64+
an event message will only ever have one of those values in the event field,
65+
and so we don't need to handle this with a more complex Pydantic schema.
66+
"""
67+
68+
69+
# This type alias is used to handle static type checking accurately while still conveying that
70+
# a value is expected to be a Literal with string type args.
71+
LiteralStrGenericAlias: TypeAlias = Annotated[ # noqa: UP040
72+
LiteralString,
73+
GetPydanticStrSchema,
74+
]
75+
"""Type alias for a generic literal string type that is compatible with Pydantic."""
76+
77+
78+
# covariant=True is used to allow subclasses of EventBase to be used in place of the base class.
79+
LiteralEventName_co = TypeVar("LiteralEventName_co", bound=PydanticLiteralStrGenericAlias, default=PydanticLiteralStrGenericAlias, covariant=True)
80+
"""Type variable for a Literal type with string args."""
81+
82+
83+
def is_literal_str_generic_alias_type(value: object | None) -> TypeGuard[LiteralStrGenericAlias]:
84+
"""Check if a type is a concrete Literal type with string args."""
3885
if value is None:
3986
return False
4087

41-
event_field_base_type = getattr(value, "__origin__", None)
88+
if isinstance(value, TypeVar):
89+
return False
90+
91+
event_field_base_type = get_origin(value)
4292

4393
if event_field_base_type is not Literal:
4494
return False
@@ -48,12 +98,10 @@ def is_literal_str_type(value: object | None) -> TypeIs[LiteralString]:
4898

4999
## EventBase implementation model of the Stream Deck Plugin SDK events.
50100

51-
class EventBase(ConfiguredBaseModel, ABC):
101+
class EventBase(ConfiguredBaseModel, ABC, Generic[LiteralEventName_co]):
52102
"""Base class for event models that represent Stream Deck Plugin SDK events."""
53-
# Configure to use the docstrings of the fields as the field descriptions.
54-
model_config = ConfigDict(use_attribute_docstrings=True, serialize_by_alias=True)
55103

56-
event: str
104+
event: LiteralEventName_co
57105
"""Name of the event used to identify what occurred.
58106
59107
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:
63111
"""Validate that the event field is a Literal[str] type."""
64112
super().__init_subclass__(**kwargs)
65113

66-
model_event_type = cls.get_event_type_annotations()
114+
# This is a GenericAlias (likely used in the subclass definition, i.e. `class ConcreteEvent(EventBase[Literal["event_name"]]):`) which is technically a subclass.
115+
# We can safely ignore this case, as we only want to validate the concrete subclass itself (`ConscreteEvent`).
116+
if get_model_origin(cls) is None:
117+
return
118+
119+
model_event_type = cls.__event_type__()
67120

68-
if not is_literal_str_type(model_event_type):
121+
if not is_literal_str_generic_alias_type(model_event_type):
69122
msg = f"The event field annotation must be a Literal[str] type. Given type: {model_event_type}"
70123
raise TypeError(msg)
71124

72125
@classmethod
73-
def get_event_type_annotations(cls) -> type[object]:
126+
def __event_type__(cls) -> type[object]:
74127
"""Get the type annotations of the subclass model's event field."""
75-
return get_type_hints(cls)["event"]
128+
return cls.model_fields["event"].annotation # type: ignore[index]
76129

77130
@classmethod
78-
def get_model_event_name(cls) -> tuple[str, ...]:
131+
def get_model_event_names(cls) -> tuple[str, ...]:
79132
"""Get the value of the subclass model's event field Literal annotation."""
80-
model_event_type = cls.get_event_type_annotations()
133+
model_event_type = cls.__event_type__()
81134

82135
# Ensure that the event field annotation is a Literal type.
83-
if not is_literal_str_type(model_event_type):
84-
msg = "The `event` field annotation of an Event model must be a Literal[str] type."
136+
if not is_literal_str_generic_alias_type(model_event_type):
137+
msg = f"The event field annotation of an Event model must be a Literal[str] type. Given type: {model_event_type}"
85138
raise TypeError(msg)
86139

87140
return get_args(model_event_type)

streamdeck/models/events/deep_link.py

+1-2
Original file line numberDiff line numberDiff line change
@@ -9,13 +9,12 @@ class DeepLinkPayload(ConfiguredBaseModel):
99
"""The deep-link URL, with the prefix omitted."""
1010

1111

12-
class DidReceiveDeepLink(EventBase):
12+
class DidReceiveDeepLink(EventBase[Literal["didReceiveDeepLink"]]):
1313
"""Occurs when Stream Deck receives a deep-link message intended for the plugin.
1414
1515
The message is re-routed to the plugin, and provided as part of the payload.
1616
One-way deep-link message can be routed to the plugin using the URL format:
1717
streamdeck://plugins/message/<PLUGIN_UUID>/{MESSAGE}.
1818
"""
19-
event: Literal["didReceiveDeepLink"] # type: ignore[override]
2019
payload: DeepLinkPayload
2120
"""Payload containing information about the URL that triggered the event."""

streamdeck/models/events/devices.py

+2-4
Original file line numberDiff line numberDiff line change
@@ -78,13 +78,11 @@ def __repr__(self) -> str:
7878
return f"DeviceInfo(name={self.name}, type={self.type}, size={self.size})"
7979

8080

81-
class DeviceDidConnect(EventBase, DeviceSpecificEventMixin):
81+
class DeviceDidConnect(EventBase[Literal["deviceDidConnect"]], DeviceSpecificEventMixin):
8282
"""Occurs when a Stream Deck device is connected."""
83-
event: Literal["deviceDidConnect"] # type: ignore[override]
8483
device_info: Annotated[DeviceInfo, Field(alias="deviceInfo")]
8584
"""Information about the newly connected device."""
8685

8786

88-
class DeviceDidDisconnect(EventBase, DeviceSpecificEventMixin):
87+
class DeviceDidDisconnect(EventBase[Literal["deviceDidDisconnect"]], DeviceSpecificEventMixin):
8988
"""Occurs when a Stream Deck device is disconnected."""
90-
event: Literal["deviceDidDisconnect"] # type: ignore[override]

streamdeck/models/events/dials.py

+3-6
Original file line numberDiff line numberDiff line change
@@ -28,22 +28,19 @@ class DialRotatePayload(EncoderPayload):
2828

2929
## Event models for DialDown, DialRotate, and DialUp events
3030

31-
class DialDown(EventBase, ContextualEventMixin, DeviceSpecificEventMixin):
31+
class DialDown(EventBase[Literal["dialDown"]], ContextualEventMixin, DeviceSpecificEventMixin):
3232
"""Occurs when the user presses a dial (Stream Deck +)."""
33-
event: Literal["dialDown"] # type: ignore[override]
3433
payload: EncoderPayload
3534
"""Contextualized information for this event."""
3635

3736

38-
class DialRotate(EventBase, ContextualEventMixin, DeviceSpecificEventMixin):
37+
class DialRotate(EventBase[Literal["dialRotate"]], ContextualEventMixin, DeviceSpecificEventMixin):
3938
"""Occurs when the user rotates a dial (Stream Deck +)."""
40-
event: Literal["dialRotate"] # type: ignore[override]
4139
payload: DialRotatePayload
4240
"""Contextualized information for this event."""
4341

4442

45-
class DialUp(EventBase, ContextualEventMixin, DeviceSpecificEventMixin):
43+
class DialUp(EventBase[Literal["dialUp"]], ContextualEventMixin, DeviceSpecificEventMixin):
4644
"""Occurs when the user releases a pressed dial (Stream Deck +)."""
47-
event: Literal["dialUp"] # type: ignore[override]
4845
payload: EncoderPayload
4946
"""Contextualized information for this event."""

0 commit comments

Comments
 (0)