Merge pull request #85 from graphras-com/feature/typed-domain-accessors

refactor: introduce typed SceneAccessor and TimerAccessor

Author: Faerkeren

src/haclient/api.py

@@ -128,7 +128,8 @@ def __init__(
         active = self._select_active_domains(domains)
         self._accessors: dict[str, DomainAccessor[Any]] = {}
         for spec in active:
-            accessor: DomainAccessor[Any] = DomainAccessor(spec, self._factory)
+            cls = spec.accessor_cls if spec.accessor_cls is not None else DomainAccessor
+            accessor: DomainAccessor[Any] = cls(spec, self._factory)
             self._accessors[spec.accessor_name()] = accessor
             self._accessors[spec.name] = accessor
             for event_type in spec.event_subscriptions:

src/haclient/core/plugins.py

@@ -8,7 +8,13 @@
 ``ha.light`` or ``ha.scene``). It provides:
 
 * ``__call__(name)`` and ``__getitem__(name)`` for entity lookup.
-* Domain-level operations registered by the spec via ``operations``.
+* Domain-level operations registered by the spec via ``operations`` (legacy
+  third-party path) **or** via typed subclass methods (preferred path).
+
+Domains with collection-level operations should subclass `DomainAccessor`
+and register the subclass via ``DomainSpec.accessor_cls``.  This keeps the
+public API statically typed without requiring ``# type: ignore`` workarounds
+or private ``_factory`` access from outside the accessor.
 
 Third-party plugins can ship additional domains by exposing an entry
 point under the ``haclient.domains`` group; see
@@ -63,10 +69,14 @@ class DomainSpec(Generic[E]):
     on_event : callable or None
         Per-domain event handler (see `DomainEventHandler`).
     operations : dict
-        Domain-level async operations registered on the
-        `DomainAccessor`. Each value is an async callable; it will be
-        bound to the accessor so the first positional argument *is* the
-        accessor instance.
+        Legacy dynamic operation dict kept for third-party plugin
+        compatibility. Built-in domains with collection-level operations
+        should prefer ``accessor_cls`` instead.
+    accessor_cls : type[DomainAccessor] or None
+        Optional typed `DomainAccessor` subclass to instantiate for this
+        domain. When provided, the ``HAClient`` uses this class rather than
+        the base `DomainAccessor`, exposing properly typed collection-level
+        methods (e.g. ``SceneAccessor.create``).
     """
 
     name: str
@@ -75,6 +85,7 @@ class DomainSpec(Generic[E]):
     event_subscriptions: tuple[str, ...] = ()
     on_event: DomainEventHandler | None = None
     operations: dict[str, Callable[..., Any]] = field(default_factory=dict)
+    accessor_cls: type[DomainAccessor[Any]] | None = None
 
     def accessor_name(self) -> str:
         """Return the accessor attribute name (defaults to ``name``)."""
@@ -87,14 +98,14 @@ class DomainAccessor(Generic[E]):
     Returned by ``HAClient.<accessor>``. Exposes:
 
     * Lookup by short name: ``ha.light("kitchen")`` or ``ha.light["kitchen"]``.
-    * Domain-level operations registered on the spec, bound to this accessor:
-      ``await ha.scene.create("romantic", ...)``.
+    * Domain-level operations either via typed subclass methods (preferred) or
+      via legacy dynamic binding of ``spec.operations`` entries.
 
     Parameters
     ----------
     spec : DomainSpec
         The spec describing this domain.
-    factory : EntityFactory
+    factory : EntityFactoryProtocol
         Factory used to create entity instances on demand.
     """
 
@@ -103,13 +114,23 @@ def __init__(self, spec: DomainSpec[E], factory: EntityFactoryProtocol) -> None:
         self._factory = factory
         for op_name, op in spec.operations.items():
             # Bind each operation as an attribute on the instance.
+            # This path is kept for backward-compatible third-party plugins.
             setattr(self, op_name, self._bind(op))
 
     @property
     def spec(self) -> DomainSpec[E]:
         """Return the underlying `DomainSpec`."""
         return self._spec
 
+    @property
+    def factory(self) -> EntityFactoryProtocol:
+        """Return the `EntityFactoryProtocol` used to create entities.
+
+        Subclasses use this to access ``factory.services`` and
+        ``factory.state`` without reaching into private internals.
+        """
+        return self._factory
+
     def _bind(self, op: Callable[..., Any]) -> Callable[..., Any]:
         """Bind a domain operation to this accessor.
 

src/haclient/domains/scene.py

@@ -6,28 +6,24 @@
 Domain-level operations
 -----------------------
 Beyond per-entity actions, the scene domain exposes two collection-level
-operations on the `DomainAccessor`:
+operations on the `SceneAccessor` (returned by ``ha.scene``):
 
-* ``create(scene_id, entities, *, snapshot_entities=None) -> Scene`` —
-  create (or update) a runtime scene helper.
-* ``apply(entities, *, transition=None) -> None`` — apply a state
-  combination without persisting it.
+* ``await ha.scene.create(scene_id, entities, *, snapshot_entities=None)``
+  — create (or update) a runtime scene helper, returning a `Scene`.
+* ``await ha.scene.apply(entities, *, transition=None)``
+  — apply a state combination without persisting it.
 
-These are invoked as ``await ha.scene.create(...)`` and
-``await ha.scene.apply(...)``. Per-entity access still works through the
-usual ``ha.scene("name")`` / ``ha.scene["name"]`` syntax.
+Per-entity access still works through the usual
+``ha.scene("name")`` / ``ha.scene["name"]`` syntax.
 """
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+from typing import Any
 
 from haclient.core.plugins import DomainAccessor, DomainSpec, register_domain
 from haclient.entity.base import Entity, ValueChangeHandler
 
-if TYPE_CHECKING:
-    from haclient.core.factory import EntityFactory
-
 
 class Scene(Entity):
     """A Home Assistant scene entity.
@@ -108,74 +104,82 @@ def on_activate(self, func: ValueChangeHandler) -> ValueChangeHandler:
         return self._register_state_value_listener(func)
 
 
-# -- Domain-level operations --------------------------------------------
-
-
-async def _create(
-    accessor: DomainAccessor[Scene],
-    scene_id: str,
-    entities: dict[str, dict[str, Any]],
-    *,
-    snapshot_entities: list[str] | None = None,
-) -> Scene:
-    """Create a runtime scene and return the corresponding `Scene`.
-
-    Parameters
-    ----------
-    accessor : DomainAccessor
-        The scene accessor (provided automatically by the binding).
-    scene_id : str
-        Object-id for the new scene (e.g. ``"romantic"`` →
-        ``scene.romantic``).
-    entities : dict
-        Mapping of entity ids to state/attribute dicts.
-    snapshot_entities : list of str or None, optional
-        Entity ids whose current state should be captured.
-
-    Returns
-    -------
-    Scene
-        The newly created scene.
-    """
-    factory: EntityFactory = accessor._factory  # type: ignore[assignment]
-    services = factory.services
-    payload: dict[str, Any] = {"scene_id": scene_id, "entities": entities}
-    if snapshot_entities is not None:
-        payload["snapshot_entities"] = snapshot_entities
-    await services.call("scene", "create", payload)
-    return accessor[scene_id]
-
-
-async def _apply(
-    accessor: DomainAccessor[Scene],
-    entities: dict[str, dict[str, Any]],
-    *,
-    transition: float | None = None,
-) -> None:
-    """Apply a scene-like state combination without persisting it.
-
-    Parameters
-    ----------
-    accessor : DomainAccessor
-        The scene accessor.
-    entities : dict
-        Mapping of entity ids to desired state/attribute dicts.
-    transition : float or None, optional
-        Transition seconds for entities that support it.
+# -- Typed domain accessor ----------------------------------------------
+
+
+class SceneAccessor(DomainAccessor[Scene]):
+    """Typed domain accessor for the ``scene`` domain.
+
+    Returned by ``ha.scene``. Provides statically-typed collection-level
+    operations in addition to the standard entity lookup methods inherited
+    from `DomainAccessor`.
     """
-    factory: EntityFactory = accessor._factory  # type: ignore[assignment]
-    services = factory.services
-    payload: dict[str, Any] = {"entities": entities}
-    if transition is not None:
-        payload["transition"] = transition
-    await services.call("scene", "apply", payload)
+
+    async def create(
+        self,
+        scene_id: str,
+        entities: dict[str, dict[str, Any]],
+        *,
+        snapshot_entities: list[str] | None = None,
+    ) -> Scene:
+        """Create (or update) a runtime scene helper.
+
+        Parameters
+        ----------
+        scene_id : str
+            Object-id for the new scene (e.g. ``"romantic"`` →
+            ``scene.romantic``).
+        entities : dict
+            Mapping of entity ids to target state/attribute dicts.
+        snapshot_entities : list of str or None, optional
+            Entity ids whose current state should be captured instead of
+            providing an explicit state dict.
+
+        Returns
+        -------
+        Scene
+            The newly created (or updated) scene entity.
+        """
+        from haclient.core.factory import EntityFactory
+
+        factory = self.factory
+        assert isinstance(factory, EntityFactory)
+        payload: dict[str, Any] = {"scene_id": scene_id, "entities": entities}
+        if snapshot_entities is not None:
+            payload["snapshot_entities"] = snapshot_entities
+        await factory.services.call("scene", "create", payload)
+        return self[scene_id]
+
+    async def apply(
+        self,
+        entities: dict[str, dict[str, Any]],
+        *,
+        transition: float | None = None,
+    ) -> None:
+        """Apply a scene-like state combination without persisting it.
+
+        Parameters
+        ----------
+        entities : dict
+            Mapping of entity ids to desired state/attribute dicts.
+        transition : float or None, optional
+            Transition seconds for entities that support it.
+        """
+        from haclient.core.factory import EntityFactory
+
+        factory = self.factory
+        assert isinstance(factory, EntityFactory)
+        payload: dict[str, Any] = {"entities": entities}
+        if transition is not None:
+            payload["transition"] = transition
+        await factory.services.call("scene", "apply", payload)
 
 
 SPEC: DomainSpec[Scene] = register_domain(
     DomainSpec(
         name="scene",
         entity_cls=Scene,
-        operations={"create": _create, "apply": _apply},
+        accessor_cls=SceneAccessor,
     )
 )
 """The `DomainSpec` registered with the shared `DomainRegistry`."""