Skip to content

Entity Factory

factory

EntityFactory — instantiates entities lazily for a given DomainSpec.

The factory binds the four "narrow ports" each entity needs (services, state store, clock, plus the spec-defined event subscription) and short-circuits to the existing instance when one is already registered.

Type-conflict checks (e.g. trying to obtain a Light for an entity id already registered as a Switch) raise HAClientError.

EntityFactory

Bases: EntityFactoryProtocol

Create or fetch entities according to a DomainSpec.

Source code in src/haclient/core/factory.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class EntityFactory(EntityFactoryProtocol):
    """Create or fetch entities according to a `DomainSpec`."""

    def __init__(
        self,
        services: ServiceCaller,
        state: StateStore,
        clock: Clock,
    ) -> None:
        self._services = services
        self._state = state
        self._clock = clock

    @property
    def services(self) -> ServiceCaller:
        """Return the bound `ServiceCaller`."""
        return self._services

    @property
    def state(self) -> StateStore:
        """Return the bound `StateStore`."""
        return self._state

    @property
    def clock(self) -> Clock:
        """Return the bound `Clock`."""
        return self._clock

    def get_or_create(self, spec: DomainSpec[Any], name: str) -> Any:
        """Return the entity for ``spec.name + '.' + name``.

        Parameters
        ----------
        spec : DomainSpec
            The spec describing the domain.
        name : str
            Short object-id (no dot allowed).

        Returns
        -------
        Entity
            The existing or newly created entity.

        Raises
        ------
        HAClientError
            If an entity with the resolved id already exists but is
            registered with a different class than ``spec.entity_cls``.
        """
        entity_id = self._state.registry.resolve(spec.name, name)
        existing = self._state.registry.get(entity_id)
        if existing is not None:
            if not isinstance(existing, spec.entity_cls):
                raise HAClientError(
                    f"Entity {entity_id} is registered as "
                    f"{type(existing).__name__}, not {spec.entity_cls.__name__}"
                )
            return existing
        entity = spec.entity_cls(entity_id, self._services, self._state, self._clock)
        return entity

    def in_domain(self, spec: DomainSpec[Any]) -> list[Entity]:
        """Return all currently registered entities for *spec*."""
        return self._state.registry.in_domain(spec.name)

services property 

services: ServiceCaller

Return the bound ServiceCaller.

state property 

state: StateStore

Return the bound StateStore.

clock property 

clock: Clock

Return the bound Clock.

get_or_create 

get_or_create(spec: DomainSpec[Any], name: str) -> Any

Return the entity for spec.name + '.' + name.

Parameters:

Name Type Description Default
spec DomainSpec

The spec describing the domain.

 required
name str

Short object-id (no dot allowed).

 required

Returns:

Type Description
Entity

The existing or newly created entity.

Raises:

Type Description
HAClientError

If an entity with the resolved id already exists but is registered with a different class than spec.entity_cls.
Source code in src/haclient/core/factory.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def get_or_create(self, spec: DomainSpec[Any], name: str) -> Any:
    """Return the entity for ``spec.name + '.' + name``.

    Parameters
    ----------
    spec : DomainSpec
        The spec describing the domain.
    name : str
        Short object-id (no dot allowed).

    Returns
    -------
    Entity
        The existing or newly created entity.

    Raises
    ------
    HAClientError
        If an entity with the resolved id already exists but is
        registered with a different class than ``spec.entity_cls``.
    """
    entity_id = self._state.registry.resolve(spec.name, name)
    existing = self._state.registry.get(entity_id)
    if existing is not None:
        if not isinstance(existing, spec.entity_cls):
            raise HAClientError(
                f"Entity {entity_id} is registered as "
                f"{type(existing).__name__}, not {spec.entity_cls.__name__}"
            )
        return existing
    entity = spec.entity_cls(entity_id, self._services, self._state, self._clock)
    return entity

in_domain 

in_domain(spec: DomainSpec[Any]) -> list[Entity]

Return all currently registered entities for spec.

Source code in src/haclient/core/factory.py 
89
90
91
def in_domain(self, spec: DomainSpec[Any]) -> list[Entity]:
    """Return all currently registered entities for *spec*."""
    return self._state.registry.in_domain(spec.name)