Add scope-aware injectable handler parameters

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
 

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

streamdeck/__main__.py

@@ -74,9 +74,15 @@ 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(
+            name,
+            param_config.resolver,
+            scope=param_config.scope,
+        )
+
     # Register custom event types before validating action handlers.
     for listener_type in config.event_listeners:
         manager.register_event_listener(listener_type())

streamdeck/event_handlers/injectables.py

@@ -3,61 +3,135 @@
 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,
     EventModel_contra,
     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():
             if parameter.kind not in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY):
                 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.

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

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