From 96c942a34dbde2a36dc804492d3f20203f8e502d Mon Sep 17 00:00:00 2001 From: strohganoff Date: Tue, 10 Nov 2026 17:47:27 -0700 Subject: [PATCH 1/5] Add scope-aware injectable handler params --- CONFIGURATION.md | 8 ++ streamdeck/__main__.py | 3 + streamdeck/event_handlers/injectables.py | 110 +++++++++++++++--- streamdeck/event_handlers/registry.py | 48 +++++--- streamdeck/event_handlers/scopes.py | 83 +++++++++++++ streamdeck/manager.py | 40 +++++-- streamdeck/models/configs.py | 12 ++ streamdeck/models/events/adapter.py | 8 ++ .../test_command_sender_binding.py | 30 +++-- tests/plugin_manager/test_plugin_manager.py | 4 +- tests/test_launch_settings.py | 2 +- 11 files changed, 296 insertions(+), 52 deletions(-) create mode 100644 streamdeck/event_handlers/scopes.py diff --git a/CONFIGURATION.md b/CONFIGURATION.md index a46aad2..b3bdd3e 100644 --- a/CONFIGURATION.md +++ b/CONFIGURATION.md @@ -23,8 +23,16 @@ module overrides. [tool.streamdeck] action_scripts = ["myplugin.actions"] event_listener_modules = ["myplugin.listeners"] +injectable_handler_params = { current_user = {}, instance_data = { scope = "ActionInstance" } } ``` +`injectable_handler_params` maps injectable parameter names to their scope +availability. Scope names are `Plugin`, `Device`, `ActionInstance`, and +`ActionInstanceWithDevice`; an omitted scope defaults to `Plugin`, which is +available to every event. The scope controls which event models may provide a +parameter; resolver callables are registered separately in code. The built-in +`command_sender` injectable is registered by the SDK at `Plugin` scope. + ```python from streamdeck.models.configs import StreamDeckToolConfig diff --git a/streamdeck/__main__.py b/streamdeck/__main__.py index a7e15b5..cf5831a 100644 --- a/streamdeck/__main__.py +++ b/streamdeck/__main__.py @@ -77,6 +77,9 @@ def run_plugin(launch: LaunchArguments, config: StreamDeckToolConfig) -> None: info=launch.info.model_dump(), ) + for name, param_config in config.injectable_handler_params.items(): + manager.register_injectable_handler_param_scope(name, param_config.scope) + # Register custom event types before validating action handlers. for listener_type in config.event_listeners: manager.register_event_listener(listener_type()) diff --git a/streamdeck/event_handlers/injectables.py b/streamdeck/event_handlers/injectables.py index 2b62d5d..3c55610 100644 --- a/streamdeck/event_handlers/injectables.py +++ b/streamdeck/event_handlers/injectables.py @@ -3,7 +3,8 @@ from __future__ import annotations import inspect -from collections.abc import Callable, Iterable +from dataclasses import dataclass +from typing import TYPE_CHECKING, Generic, Protocol, TypeVar from streamdeck.event_handlers.protocol import ( EventHandlerFunc, @@ -11,46 +12,111 @@ inspect_event_handler_signature, validate_event_handler_event_parameter, ) +from streamdeck.event_handlers.scopes import Scope, ScopeName, scope_covers + + +if TYPE_CHECKING: + from collections.abc import Callable, Iterable + + from streamdeck.models.events.base import EventBase + + +# Resolvers and specs only expose their value as a return type, so this can be covariant. +Value_co = TypeVar("Value_co", covariant=True) +"""Covariant injectable value returned by a resolver.""" + + +# `register` accepts a resolver, so its method-level type variable must remain invariant. +Value = TypeVar("Value") +"""Injectable value type inferred when registering a resolver.""" + + +class InjectableResolver(Protocol[Value_co]): + """Resolve an injectable value using the event being handled.""" + + def __call__(self, event_data: EventBase, /) -> Value_co: + """Return the injectable value for an event.""" + ... + + +@dataclass(frozen=True) +class InjectableSpec(Generic[Value_co]): + """Resolver and availability scope for an injectable handler parameter.""" + + scope: ScopeName + """Narrowest event scope where this injectable is available.""" + + resolve: InjectableResolver[Value_co] + """Resolve the injectable value for the event being handled.""" class InjectableHandlerParamStore: - """Store injectable objects and resolve them by handler parameter name.""" + """Store event-aware injectable resolvers and resolve them by handler parameter name.""" def __init__(self) -> None: - self._values: dict[str, object] = {} + """Initialize an empty injectable resolver store.""" + self._specs: dict[str, InjectableSpec[object]] = {} + self._scopes: dict[str, ScopeName] = {} @property def parameter_names(self) -> frozenset[str]: - """Names of the objects currently held by the store.""" - return frozenset(self._values) + """Names of the injectable resolvers currently held by the store.""" + return frozenset(self._specs) + + def register_scope(self, name: str, scope: ScopeName) -> None: + """Set the availability scope for an injectable handler parameter.""" + if name == "event_data": + msg = "'event_data' is reserved for the required event parameter" + raise ValueError(msg) + + self._scopes[name] = scope + if name in self._specs: + spec = self._specs[name] + self._specs[name] = InjectableSpec(scope=scope, resolve=spec.resolve) - def set_value(self, name: str, value: object) -> None: - """Store a value under an injectable parameter name.""" + def register( + self, + name: str, + resolver: InjectableResolver[Value], + *, + scope: ScopeName | None = None, + ) -> None: + """Register a resolver under an injectable parameter name.""" if name == "event_data": msg = "'event_data' is reserved for the required event parameter" raise ValueError(msg) - self._values[name] = value + spec: InjectableSpec[Value] = InjectableSpec( + scope=scope if scope is not None else self._scopes.get(name, "Plugin"), + resolve=resolver, + ) + self._specs[name] = spec - def resolve(self, parameter_names: Iterable[str]) -> dict[str, object]: - """Return stored values whose names are requested by a handler.""" + def resolve(self, parameter_names: Iterable[str], event_data: EventBase) -> dict[str, object]: + """Resolve named injectables for the event being handled.""" return { - name: self._values[name] + name: self._specs[name].resolve(event_data) for name in parameter_names - if name in self._values + if name in self._specs } - def resolve_for_handler(self, handler: Callable[..., None]) -> dict[str, object]: - """Return stored values only for parameters declared by a handler.""" - return self.resolve(inspect.signature(handler).parameters) + def resolve_for_handler(self, handler: Callable[..., None], event_data: EventBase) -> dict[str, object]: + """Resolve values only for injectable parameters declared by a handler.""" + return self.resolve(inspect.signature(handler).parameters, event_data) - def validate_handler_signature(self, handler: EventHandlerFunc[EventModel_contra, ...]) -> None: - """Validate that the store can supply every required handler parameter after the event.""" + def validate_handler_signature( + self, + handler: EventHandlerFunc[EventModel_contra, ...], + event_scopes: Iterable[Scope], + event_name: str, + ) -> None: + """Validate handler binding and injectable scopes for one subscribed event.""" handler_signature = inspect_event_handler_signature(handler) validate_event_handler_event_parameter(handler_signature) + event_scopes = tuple(event_scopes) injectable_parameters = { name: handler_signature.parameters[name] - for name in self.parameter_names + for name in self._specs if name in handler_signature.parameters } for name, parameter in injectable_parameters.items(): @@ -58,6 +124,14 @@ def validate_handler_signature(self, handler: EventHandlerFunc[EventModel_contra msg = f"Invalid event handler signature for {handler!r}: {name!r} must accept a keyword argument" raise TypeError(msg) + spec = self._specs[name] + if any(not scope_covers(spec.scope, event_scope) for event_scope in event_scopes): + msg = ( + f"Invalid event handler signature for {handler!r}: injectable parameter {name!r} " + f"requires {spec.scope!r} scope, which is not available for event {event_name!r}" + ) + raise TypeError(msg) + # Check that dispatch can satisfy every required parameter without calling the handler. # Simulate dispatch's call shape: event_data positionally, plus stored values by keyword. # Binding fails if any required parameter is missing. diff --git a/streamdeck/event_handlers/registry.py b/streamdeck/event_handlers/registry.py index 6147ec8..46de290 100644 --- a/streamdeck/event_handlers/registry.py +++ b/streamdeck/event_handlers/registry.py @@ -1,6 +1,9 @@ +"""Registry for action and plugin event handler catalogs.""" + from __future__ import annotations from typing import TYPE_CHECKING + from typing_extensions import sentinel from streamdeck.event_handlers.actions import Action @@ -34,12 +37,40 @@ def register(self, catalog: SupportsEventHandlers) -> None: """ self._plugin_event_handler_catalogs.append(catalog) + def iter_event_handler_bindings( + self, + event_name: EventNameStr | ALL_EVENTS = ALL_EVENTS, + event_action_uuid: ActionUUIDStr | None = None, + ) -> Iterator[tuple[EventNameStr, EventHandlerFunc]]: + """Yield registered event names with their handlers, optionally filtered. + + Args: + event_name: The event name to retrieve, or ``ALL_EVENTS`` to enumerate every + registered event. + event_action_uuid: The action UUID to filter action-specific catalogs by. If None, + include handlers from all catalogs. + + Yields: + A registered event name and its handler. + """ + for catalog in self._plugin_event_handler_catalogs: + if event_action_uuid is not None and (isinstance(catalog, Action) and catalog.uuid != event_action_uuid): + continue + + for catalog_event_name in ( + (event_name,) + if event_name is not ALL_EVENTS + else catalog.get_registered_event_names() + ): + for handler in catalog.get_event_handlers(catalog_event_name): + yield catalog_event_name, handler + def get_event_handlers( self, event_name: EventNameStr | ALL_EVENTS, event_action_uuid: ActionUUIDStr | None = None, ) -> Iterator[EventHandlerFunc]: - """Get handlers matching an event name or enumerate all registrations. + """Yield handlers matching an event name or enumerate all registrations. Passing ``ALL_EVENTS`` enumerates handlers registered under every event name. It does not register a handler to receive every event. @@ -53,16 +84,5 @@ def get_event_handlers( Yields: EventHandlerFunc: The event handler functions for the specified event. """ - for catalog in self._plugin_event_handler_catalogs: - # If the event is action-specific (i.e is not a GlobalAction and has a UUID attribute), - # only get handlers for that action, as we don't want to trigger - # and pass this event to handlers for other actions. - if event_action_uuid is not None and (isinstance(catalog, Action) and catalog.uuid != event_action_uuid): - continue - - for catalog_event_name in ( - (event_name,) - if event_name is not ALL_EVENTS - else catalog.get_registered_event_names() - ): - yield from catalog.get_event_handlers(catalog_event_name) + for _, handler in self.iter_event_handler_bindings(event_name, event_action_uuid): + yield handler diff --git a/streamdeck/event_handlers/scopes.py b/streamdeck/event_handlers/scopes.py new file mode 100644 index 0000000..6cd6475 --- /dev/null +++ b/streamdeck/event_handlers/scopes.py @@ -0,0 +1,83 @@ +"""Scopes used to limit which events can provide injectable parameters.""" + +from __future__ import annotations + +from enum import Flag, auto +from typing import TYPE_CHECKING, Literal + +from streamdeck.models.events.common import ContextualEventMixin, DeviceSpecificEventMixin + + +if TYPE_CHECKING: + from collections.abc import Callable + + from streamdeck.models.events.base import EventBase + + +ScopeName = Literal["Plugin", "Device", "ActionInstance", "ActionInstanceWithDevice"] +"""A named injectable scope exposed in plugin configuration.""" + + +class Scope(Flag): + """Represent event identity capabilities used to check injectable availability. + + Flags can be combined when an event provides multiple identities. `PLUGIN` is + zero because plugin-scoped injectables require no event identity. + """ + + PLUGIN = 0 + """No device or action-instance identity is required.""" + DEVICE = auto() + """The event provides a device identity.""" + ACTION_INSTANCE = auto() + """The event provides an action-instance identity.""" + + ACTION_INSTANCE_WITH_DEVICE = DEVICE | ACTION_INSTANCE + """The event provides both device and action-instance identities.""" + + +SCOPE_CAPABILITIES: dict[ScopeName, Scope] = { + "Plugin": Scope.PLUGIN, + "Device": Scope.DEVICE, + "ActionInstance": Scope.ACTION_INSTANCE, + "ActionInstanceWithDevice": Scope.ACTION_INSTANCE_WITH_DEVICE, +} +"""Map each configured scope name to the identities it requires.""" + + +def scope_covers(required_scope: ScopeName, event_scope: Scope) -> bool: + """Return whether an event scope provides the identities required by an injectable. + + Args: + required_scope: The minimum scope required by the injectable parameter. + event_scope: The identity scope available on the event. + + Returns: + Whether the event provides every identity required by the injectable. + """ + required = SCOPE_CAPABILITIES[required_scope] + return required & event_scope == required + + +def scope_for_event(event_or_model: EventBase | type[EventBase]) -> Scope: + """Return the identity capabilities available on an event or its model. + + Args: + event_or_model: A validated event instance or its event model class. + + Returns: + The combined device and action-instance identities provided by the event. + """ + # Params default to Plugin scope (no filtering is applied) + scope = Scope.PLUGIN + + scope_check: Callable[..., bool] = ( + issubclass if isinstance(event_or_model, type) else isinstance + ) + + if scope_check(event_or_model, ContextualEventMixin): + scope |= Scope.ACTION_INSTANCE + if scope_check(event_or_model, DeviceSpecificEventMixin): + scope |= Scope.DEVICE + + return scope diff --git a/streamdeck/manager.py b/streamdeck/manager.py index 65eeb8a..df347f5 100644 --- a/streamdeck/manager.py +++ b/streamdeck/manager.py @@ -9,7 +9,8 @@ from streamdeck.event_handlers.actions import Action, ActionBase from streamdeck.event_handlers.injectables import InjectableHandlerParamStore from streamdeck.event_handlers.protocol import SupportsEventHandlers -from streamdeck.event_handlers.registry import ALL_EVENTS, HandlersRegistry +from streamdeck.event_handlers.registry import HandlersRegistry +from streamdeck.event_handlers.scopes import ScopeName, scope_for_event from streamdeck.event_listener import EventListener, EventListenerManager from streamdeck.models.events.adapter import EventAdapter from streamdeck.models.events.common import ContextualEventMixin @@ -21,6 +22,7 @@ from collections.abc import Iterator from typing import Any, Literal + from streamdeck.event_handlers.injectables import InjectableResolver from streamdeck.models.events import EventBase @@ -38,7 +40,7 @@ def __init__( plugin_uuid: str, plugin_registration_uuid: str, # Passed in by Streamdeck to the entry-point script. should we compare what's in the manifest.json file? register_event: Literal["registerPlugin"], # Passed in by Streamdeck to the entry-point script. Will this always be "registerPlugin"? - info: dict[str, Any] + info: dict[str, Any], ): """Initialize a PluginManager instance. @@ -95,9 +97,17 @@ def register_action(self, action: ActionBase) -> None: self._handlers_registry.register(action) def _validate_registered_handler_signatures(self) -> None: - """Ensure registered handlers can be called with the current injectable values.""" - for handler in self._handlers_registry.get_event_handlers(event_name=ALL_EVENTS): - self._injectable_handler_param_store.validate_handler_signature(handler) + """Ensure handlers can bind and resolve their injectables for every subscribed event.""" + for event_name, handler in self._handlers_registry.iter_event_handler_bindings(): + event_scopes = ( + scope_for_event(model) + for model in self._event_adapter.models_for_event_name(event_name) + ) + self._injectable_handler_param_store.validate_handler_signature( + handler, + event_scopes, + event_name, + ) def register_event_listener(self, listener: EventListener) -> None: """Register an event listener with the PluginManager, and add its event models to the event adapter. @@ -110,6 +120,18 @@ def register_event_listener(self, listener: EventListener) -> None: for event_model in listener.event_models: self._event_adapter.add_model(event_model) + def register_injectable_handler_param( + self, + name: str, + resolver: InjectableResolver[object], + ) -> None: + """Register a custom injectable resolver using its configured scope.""" + self._injectable_handler_param_store.register(name, resolver) + + def register_injectable_handler_param_scope(self, name: str, scope: ScopeName) -> None: + """Register the availability scope for an injectable handler parameter.""" + self._injectable_handler_param_store.register_scope(name, scope) + def _stream_event_data(self) -> Iterator[EventBase]: """Stream event data from the event listeners. @@ -143,7 +165,7 @@ def _dispatch_event(self, event: EventBase) -> None: # If the event is action-specific, we'll pass the action's uuid to the handler to ensure only the correct action is triggered. event_action_uuid = event.action if isinstance(event, ContextualEventMixin) else None for event_handler in self._handlers_registry.get_event_handlers(event_name=event.event, event_action_uuid=event_action_uuid): - injected_params = self._injectable_handler_param_store.resolve_for_handler(event_handler) + injected_params = self._injectable_handler_param_store.resolve_for_handler(event_handler, event) # TODO: from contextual event occurrences, save metadata to the action's properties. event_handler(event, **injected_params) @@ -159,7 +181,11 @@ def run(self) -> None: """ client = WebSocketClient(port=self._port) command_sender = StreamDeckCommandSender(client, plugin_registration_uuid=self._registration_uuid) - self._injectable_handler_param_store.set_value("command_sender", command_sender) + self._injectable_handler_param_store.register( + "command_sender", + lambda _event_data: command_sender, + scope="Plugin", + ) self._validate_registered_handler_signatures() with client: diff --git a/streamdeck/models/configs.py b/streamdeck/models/configs.py index d3d7db0..2f84849 100644 --- a/streamdeck/models/configs.py +++ b/streamdeck/models/configs.py @@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Annotated, Any from pydantic import ( + BaseModel, Field, ImportString, ) @@ -19,6 +20,9 @@ ) from streamdeck.event_handlers.actions import ActionBase + +# Pydantic resolves this postponed annotation at runtime. +from streamdeck.event_handlers.scopes import ScopeName # noqa: TC001 from streamdeck.event_listener import EventListener from streamdeck.models.events import EventBase @@ -52,6 +56,13 @@ def _add_plugin_module_argument( return None +class InjectableHandlerParamConfig(BaseModel): + """Configuration for an injectable handler parameter.""" + + scope: ScopeName = "Plugin" + """The minimum event scope where this injectable parameter is available.""" + + class StreamDeckToolConfig(BaseSettings): """Plugin configuration from CLI values followed by [tool.streamdeck].""" @@ -66,6 +77,7 @@ class StreamDeckToolConfig(BaseSettings): action_script_modules: Annotated[list[ImportString[ModuleType]], Field(alias="action_scripts")] event_listener_modules: list[ImportString[ModuleType]] = Field(default_factory=list) + injectable_handler_params: dict[str, InjectableHandlerParamConfig] = Field(default_factory=dict) @classmethod def settings_customise_sources( diff --git a/streamdeck/models/events/adapter.py b/streamdeck/models/events/adapter.py index ee406cb..4464171 100644 --- a/streamdeck/models/events/adapter.py +++ b/streamdeck/models/events/adapter.py @@ -39,6 +39,14 @@ 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 + def models_for_event_name(self, event_name: str) -> tuple[type[EventBase], ...]: + """Return all event models that can validate the supplied event name.""" + return tuple( + model + for model in self._models + if event_name in model.get_model_event_names() + ) + @property def type_adapter(self) -> TypeAdapter[EventBase]: """Get the TypeAdapter instance for the event models.""" diff --git a/tests/plugin_manager/test_command_sender_binding.py b/tests/plugin_manager/test_command_sender_binding.py index 1bc059e..8b34e14 100644 --- a/tests/plugin_manager/test_command_sender_binding.py +++ b/tests/plugin_manager/test_command_sender_binding.py @@ -6,6 +6,7 @@ import pytest from streamdeck.event_handlers.actions import Action from streamdeck.event_handlers.injectables import InjectableHandlerParamStore +from streamdeck.event_handlers.scopes import Scope if TYPE_CHECKING: @@ -44,9 +45,10 @@ def test_injectable_param_store_resolves_only_declared_params() -> None: handler_without_sender = create_event_handler(include_command_sender_param=False) store = InjectableHandlerParamStore() - store.set_value("command_sender", mock_command_sender) - with_sender = store.resolve_for_handler(handler_with_sender) - without_sender = store.resolve_for_handler(handler_without_sender) + store.register("command_sender", lambda _event_data: mock_command_sender) + event_data = Mock() + with_sender = store.resolve_for_handler(handler_with_sender, event_data) + without_sender = store.resolve_for_handler(handler_without_sender, event_data) assert with_sender == {"command_sender": mock_command_sender} assert without_sender == {} @@ -57,10 +59,10 @@ def test_injectable_param_store_resolves_multiple_named_values() -> None: command_sender = Mock() event_snapshot = object() store = InjectableHandlerParamStore() - store.set_value("command_sender", command_sender) - store.set_value("event_snapshot", event_snapshot) + store.register("command_sender", lambda _event_data: command_sender) + store.register("event_snapshot", lambda _event_data: event_snapshot) - injected_params = store.resolve(("command_sender", "event_snapshot", "unused")) + injected_params = store.resolve(("command_sender", "event_snapshot", "unused"), Mock()) assert injected_params == { "command_sender": command_sender, @@ -72,7 +74,10 @@ def test_manager_validates_and_resolves_custom_store_value(plugin_manager: Plugi """Registration and dispatch use the same manager-owned injectable names.""" provided_value = object() plugin_manager._injectable_handler_param_store = InjectableHandlerParamStore() - plugin_manager._injectable_handler_param_store.set_value("extension_value", provided_value) + plugin_manager._injectable_handler_param_store.register( + "extension_value", + lambda _event_data: provided_value, + ) def handler(event_data: events.EventBase, extension_value: object) -> None: pass @@ -98,10 +103,17 @@ def handler(event_data: events.EventBase, missing_value: object) -> None: action = Action("test.action") action.on("applicationDidLaunch")(handler) plugin_manager.register_action(action) - plugin_manager._injectable_handler_param_store.set_value("command_sender", Mock()) + plugin_manager._injectable_handler_param_store.register( + "command_sender", + lambda _event_data: Mock(), + ) with pytest.raises(TypeError, match="Invalid event handler signature"): - plugin_manager._injectable_handler_param_store.validate_handler_signature(handler) + plugin_manager._injectable_handler_param_store.validate_handler_signature( + handler, + (Scope.PLUGIN,), + "applicationDidLaunch", + ) @pytest.mark.usefixtures("patch_websocket_client") diff --git a/tests/plugin_manager/test_plugin_manager.py b/tests/plugin_manager/test_plugin_manager.py index 618b9bb..02119fd 100644 --- a/tests/plugin_manager/test_plugin_manager.py +++ b/tests/plugin_manager/test_plugin_manager.py @@ -4,7 +4,6 @@ import pytest import pytest_mock from streamdeck.event_handlers.actions import Action -from streamdeck.event_handlers.registry import ALL_EVENTS from streamdeck.manager import EventAdapter, PluginManager from streamdeck.models.events import ( #, event_adapter DEFAULT_EVENT_MODELS, @@ -138,10 +137,9 @@ def test_plugin_manager_process_event( # Check that the validate_json method returns the same event type model as the fake_event_message. assert spied_event_adapter__validate_json.spy_return == fake_event_message - # Startup validation enumerates all handlers before dispatch queries for this event. + # Startup validation inspects registered event bindings before dispatch queries for this event. spied_handlers_registry__get_event_handlers = cast("Mock", plugin_manager._handlers_registry.get_event_handlers) assert spied_handlers_registry__get_event_handlers.call_args_list == [ - call(event_name=ALL_EVENTS), call(event_name=fake_event_message.event, event_action_uuid=fake_event_message.action), ] diff --git a/tests/test_launch_settings.py b/tests/test_launch_settings.py index 0653db4..e2f294e 100644 --- a/tests/test_launch_settings.py +++ b/tests/test_launch_settings.py @@ -261,7 +261,7 @@ def test_main_logs_startup_and_runtime_failures(monkeypatch, caplog, phase): error = RuntimeError(f"{phase} failed") listener_type = Mock() - config = Mock(event_listeners=[listener_type], actions=[object()]) + config = Mock(event_listeners=[listener_type], actions=[object()], injectable_handler_params={}) settings_factory = Mock(return_value=config) manager = Mock() if phase == "settings": From d95e0f90a0631241b3da32ae39cc8dba63c1fef4 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Wed, 11 Nov 2026 13:39:11 -0700 Subject: [PATCH 2/5] Configure injectable resolver imports --- README.md | 20 +++++++++++++++++++- streamdeck/__main__.py | 1 + streamdeck/models/configs.py | 3 +++ 3 files changed, 23 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 6a34ac9..5f125d5 100644 --- a/README.md +++ b/README.md @@ -90,7 +90,7 @@ def handle_will_appear(event_data): !!!INFO Handlers for action-specific events are dispatched only if the event is triggered by the associated action, ensuring isolation and predictability. For other types of events that are not associated with a specific action, handlers are dispatched without such restrictions. -A handler must have a required positional parameter named `event_data` first; the `.on()` decorator checks this when applied. At startup, before opening the WebSocket, the manager checks that any other required parameters have a stored injectable value. Currently, `command_sender` is the only injected parameter. The SDK passes it by keyword, so declare it as positional-or-keyword or keyword-only. +A handler must have a required positional parameter named `event_data` first; the `.on()` decorator checks this when applied. At startup, before opening the WebSocket, the manager checks that any other required parameters have a stored injectable value. The SDK provides `command_sender`; plugins can define additional injected parameters in `pyproject.toml`. The SDK passes injectable values by keyword, so declare them as positional-or-keyword or keyword-only. ```python from streamdeck.command_sender import StreamDeckCommandSender @@ -105,6 +105,24 @@ def handle_will_appear(event_data, command_sender: StreamDeckCommandSender): The `command_sender` parameter is optional. If included in the handler's signature, the SDK automatically injects a `StreamDeckCommandSender` instance that can be used to send commands back to the Stream Deck (like setting titles, states, or images). Startup raises `TypeError` for signatures whose required parameters cannot be supplied by dispatch. +Custom injectable parameters name an importable resolver in `pyproject.toml`. The table key is the handler parameter name, and the resolver is called with the event data whenever a handler requests that parameter: + +```toml +[tool.streamdeck.injectable_handler_params.api_client] +resolver = "my_plugin.dependencies.resolve_api_client" +scope = "Plugin" +``` + +```python +# my_plugin/dependencies.py +api_client = ApiClient() + +def resolve_api_client(event_data): + return api_client +``` + +The resolver can return a module-level instance or compute an event-specific value. `scope` defaults to `Plugin` when omitted. + ### Writing Logs diff --git a/streamdeck/__main__.py b/streamdeck/__main__.py index cf5831a..2656a71 100644 --- a/streamdeck/__main__.py +++ b/streamdeck/__main__.py @@ -79,6 +79,7 @@ def run_plugin(launch: LaunchArguments, config: StreamDeckToolConfig) -> None: for name, param_config in config.injectable_handler_params.items(): manager.register_injectable_handler_param_scope(name, param_config.scope) + manager.register_injectable_handler_param(name, param_config.resolver) # Register custom event types before validating action handlers. for listener_type in config.event_listeners: diff --git a/streamdeck/models/configs.py b/streamdeck/models/configs.py index 2f84849..67a3e35 100644 --- a/streamdeck/models/configs.py +++ b/streamdeck/models/configs.py @@ -2,6 +2,7 @@ from __future__ import annotations +from collections.abc import Callable from functools import cached_property from types import ModuleType # noqa: TC003 from typing import TYPE_CHECKING, Annotated, Any @@ -59,6 +60,8 @@ def _add_plugin_module_argument( class InjectableHandlerParamConfig(BaseModel): """Configuration for an injectable handler parameter.""" + resolver: ImportString[Callable[[EventBase], object]] + """Importable resolver called with event data to provide the handler parameter value.""" scope: ScopeName = "Plugin" """The minimum event scope where this injectable parameter is available.""" From 7e04190698ac4230ea6ff80f462e732ea45d20ed Mon Sep 17 00:00:00 2001 From: strohganoff Date: Wed, 11 Nov 2026 14:09:20 -0700 Subject: [PATCH 3/5] Register injectable resolvers with their scopes --- streamdeck/__main__.py | 8 +++++--- streamdeck/manager.py | 19 ++++++++++++------- tests/plugin_manager/conftest.py | 2 -- tests/test_launch_settings.py | 1 - 4 files changed, 17 insertions(+), 13 deletions(-) diff --git a/streamdeck/__main__.py b/streamdeck/__main__.py index 2656a71..d9f455a 100644 --- a/streamdeck/__main__.py +++ b/streamdeck/__main__.py @@ -74,12 +74,14 @@ def run_plugin(launch: LaunchArguments, config: StreamDeckToolConfig) -> None: plugin_uuid=launch.info.plugin_uuid, plugin_registration_uuid=launch.plugin_registration_uuid, register_event=launch.register_event, - info=launch.info.model_dump(), ) for name, param_config in config.injectable_handler_params.items(): - manager.register_injectable_handler_param_scope(name, param_config.scope) - manager.register_injectable_handler_param(name, param_config.resolver) + manager.register_injectable_handler_param( + name, + param_config.resolver, + scope=param_config.scope, + ) # Register custom event types before validating action handlers. for listener_type in config.event_listeners: diff --git a/streamdeck/manager.py b/streamdeck/manager.py index df347f5..34312b1 100644 --- a/streamdeck/manager.py +++ b/streamdeck/manager.py @@ -20,7 +20,7 @@ if TYPE_CHECKING: from collections.abc import Iterator - from typing import Any, Literal + from typing import Literal from streamdeck.event_handlers.injectables import InjectableResolver from streamdeck.models.events import EventBase @@ -40,7 +40,6 @@ def __init__( plugin_uuid: str, plugin_registration_uuid: str, # Passed in by Streamdeck to the entry-point script. should we compare what's in the manifest.json file? register_event: Literal["registerPlugin"], # Passed in by Streamdeck to the entry-point script. Will this always be "registerPlugin"? - info: dict[str, Any], ): """Initialize a PluginManager instance. @@ -53,14 +52,11 @@ def __init__( used to send back in the registerPlugin event. register_event (str): The registration event type, passed in by the Stream Deck software as -registerEvent option. It's value will almost definitely will be "registerPlugin". - info (dict[str, Any]): The information related to the plugin. """ self._port = port self.uuid = plugin_uuid self._registration_uuid = plugin_registration_uuid self._register_event = register_event - self._info = info - self._handlers_registry = HandlersRegistry() self._event_listener_manager = EventListenerManager() self._event_adapter = EventAdapter() @@ -124,9 +120,18 @@ def register_injectable_handler_param( self, name: str, resolver: InjectableResolver[object], + *, + scope: ScopeName | None = None, ) -> None: - """Register a custom injectable resolver using its configured scope.""" - self._injectable_handler_param_store.register(name, resolver) + """Register an event-aware resolver for a handler parameter. + + Args: + name: Handler parameter name supplied by this resolver. + resolver: Callable that returns the value to inject for an event. + scope: Optional availability scope override. If omitted, a previously registered + scope or the default `Plugin` scope is used. + """ + self._injectable_handler_param_store.register(name, resolver, scope=scope) def register_injectable_handler_param_scope(self, name: str, scope: ScopeName) -> None: """Register the availability scope for an injectable handler parameter.""" diff --git a/tests/plugin_manager/conftest.py b/tests/plugin_manager/conftest.py index b77f924..3b37484 100644 --- a/tests/plugin_manager/conftest.py +++ b/tests/plugin_manager/conftest.py @@ -25,14 +25,12 @@ def plugin_manager(port_number: int, plugin_registration_uuid: str) -> PluginMan """ plugin_uuid = "test-plugin-uuid" register_event = "registerPlugin" - info = {"some": "info"} return PluginManager( port=port_number, plugin_uuid=plugin_uuid, plugin_registration_uuid=plugin_registration_uuid, register_event=register_event, - info=info, ) diff --git a/tests/test_launch_settings.py b/tests/test_launch_settings.py index e2f294e..e20c9b3 100644 --- a/tests/test_launch_settings.py +++ b/tests/test_launch_settings.py @@ -249,7 +249,6 @@ def test_launcher_imports_from_plugin_directory_and_registers_listeners_first( assert factory.call_args.kwargs["port"] == 12345 assert factory.call_args.kwargs["plugin_uuid"] == "com.example.plugin" assert factory.call_args.kwargs["plugin_registration_uuid"] == "registration-id" - assert factory.call_args.kwargs["info"] == {"plugin_uuid": "com.example.plugin"} debug.assert_called_once_with(5678) finally: sys.modules.pop("local_plugin_actions", None) From 803d25037366bda21ca4e3eb6079818c0eb18052 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Wed, 11 Nov 2026 14:46:01 -0700 Subject: [PATCH 4/5] Remove obsolete injectable scope registration --- streamdeck/manager.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/streamdeck/manager.py b/streamdeck/manager.py index 34312b1..115643d 100644 --- a/streamdeck/manager.py +++ b/streamdeck/manager.py @@ -133,10 +133,6 @@ def register_injectable_handler_param( """ self._injectable_handler_param_store.register(name, resolver, scope=scope) - def register_injectable_handler_param_scope(self, name: str, scope: ScopeName) -> None: - """Register the availability scope for an injectable handler parameter.""" - self._injectable_handler_param_store.register_scope(name, scope) - def _stream_event_data(self) -> Iterator[EventBase]: """Stream event data from the event listeners. From dc23e241aaea8f6abfd74f65c52db04925c83866 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Wed, 11 Nov 2026 15:28:08 -0700 Subject: [PATCH 5/5] Keep runtime Callable import for Pydantic annotations --- streamdeck/models/configs.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/streamdeck/models/configs.py b/streamdeck/models/configs.py index 67a3e35..6e7b253 100644 --- a/streamdeck/models/configs.py +++ b/streamdeck/models/configs.py @@ -2,7 +2,8 @@ from __future__ import annotations -from collections.abc import Callable +# Pydantic resolves this annotation at runtime, so Callable must remain a runtime import. +from collections.abc import Callable # noqa: TC003 from functools import cached_property from types import ModuleType # noqa: TC003 from typing import TYPE_CHECKING, Annotated, Any