Skip to content

Entity Registry

registry

In-memory entity registry.

The registry stores Entity instances keyed by their fully-qualified entity_id and supports lookup by short object-id scoped to a domain. It is owned by each HAClient instance — there are no global singletons.

EntityRegistry

Mapping of entity_id to Entity.

Source code in src/haclient/core/registry.py
 19
 20
 21
 22
 23
 24
 25
 26
 27
 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
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class EntityRegistry:
    """Mapping of ``entity_id`` to `Entity`."""

    def __init__(self) -> None:
        self._entities: dict[str, Entity] = {}

    def register(self, entity: Entity) -> None:
        """Register ``entity``, overwriting any existing entry."""
        self._entities[entity.entity_id] = entity

    def unregister(self, entity_id: str) -> None:
        """Remove the entity identified by ``entity_id`` if present."""
        self._entities.pop(entity_id, None)

    def get(self, entity_id: str) -> Entity | None:
        """Return the entity for ``entity_id`` or ``None`` if missing."""
        return self._entities.get(entity_id)

    def require(self, entity_id: str) -> Entity:
        """Return the entity for ``entity_id`` or raise.

        Parameters
        ----------
        entity_id : str
            Fully-qualified entity id.

        Returns
        -------
        Entity
            The registered entity.

        Raises
        ------
        EntityNotFoundError
            If no entity is registered under ``entity_id``.
        """
        entity = self._entities.get(entity_id)
        if entity is None:
            raise EntityNotFoundError(f"Entity not found: {entity_id}")
        return entity

    def __contains__(self, entity_id: object) -> bool:
        return isinstance(entity_id, str) and entity_id in self._entities

    def __iter__(self) -> Iterator[Entity]:
        return iter(self._entities.values())

    def __len__(self) -> int:
        return len(self._entities)

    def clear(self) -> None:
        """Remove all registered entities."""
        self._entities.clear()

    def resolve(self, domain: str, name: str) -> str:
        """Build a fully-qualified ``entity_id`` from *domain* and *name*.

        Parameters
        ----------
        domain : str
            The HA domain.
        name : str
            The short object-id (no dot allowed).

        Returns
        -------
        str
            The fully-qualified entity id.

        Raises
        ------
        ValueError
            If *name* contains a dot.
        """
        if "." in name:
            hint = name.split(".", 1)[1]
            raise ValueError(
                f"Pass the short object-id (e.g. {hint!r}), "
                f"not the fully-qualified entity id {name!r}"
            )
        return f"{domain}.{name}"

    def in_domain(self, domain: str) -> list[Entity]:
        """Return all entities whose id starts with ``{domain}.``."""
        prefix = f"{domain}."
        return [e for eid, e in self._entities.items() if eid.startswith(prefix)]

register

register(entity: Entity) -> None

Register entity, overwriting any existing entry.

Source code in src/haclient/core/registry.py
25
26
27
def register(self, entity: Entity) -> None:
    """Register ``entity``, overwriting any existing entry."""
    self._entities[entity.entity_id] = entity

unregister

unregister(entity_id: str) -> None

Remove the entity identified by entity_id if present.

Source code in src/haclient/core/registry.py
29
30
31
def unregister(self, entity_id: str) -> None:
    """Remove the entity identified by ``entity_id`` if present."""
    self._entities.pop(entity_id, None)

get

get(entity_id: str) -> Entity | None

Return the entity for entity_id or None if missing.

Source code in src/haclient/core/registry.py
33
34
35
def get(self, entity_id: str) -> Entity | None:
    """Return the entity for ``entity_id`` or ``None`` if missing."""
    return self._entities.get(entity_id)

require

require(entity_id: str) -> Entity

Return the entity for entity_id or raise.

Parameters:

Name Type Description Default
entity_id str

Fully-qualified entity id.

required

Returns:

Type Description
Entity

The registered entity.

Raises:

Type Description
EntityNotFoundError

If no entity is registered under entity_id.

Source code in src/haclient/core/registry.py
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def require(self, entity_id: str) -> Entity:
    """Return the entity for ``entity_id`` or raise.

    Parameters
    ----------
    entity_id : str
        Fully-qualified entity id.

    Returns
    -------
    Entity
        The registered entity.

    Raises
    ------
    EntityNotFoundError
        If no entity is registered under ``entity_id``.
    """
    entity = self._entities.get(entity_id)
    if entity is None:
        raise EntityNotFoundError(f"Entity not found: {entity_id}")
    return entity

clear

clear() -> None

Remove all registered entities.

Source code in src/haclient/core/registry.py
69
70
71
def clear(self) -> None:
    """Remove all registered entities."""
    self._entities.clear()

resolve

resolve(domain: str, name: str) -> str

Build a fully-qualified entity_id from domain and name.

Parameters:

Name Type Description Default
domain str

The HA domain.

required
name str

The short object-id (no dot allowed).

required

Returns:

Type Description
str

The fully-qualified entity id.

Raises:

Type Description
ValueError

If name contains a dot.

Source code in src/haclient/core/registry.py
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def resolve(self, domain: str, name: str) -> str:
    """Build a fully-qualified ``entity_id`` from *domain* and *name*.

    Parameters
    ----------
    domain : str
        The HA domain.
    name : str
        The short object-id (no dot allowed).

    Returns
    -------
    str
        The fully-qualified entity id.

    Raises
    ------
    ValueError
        If *name* contains a dot.
    """
    if "." in name:
        hint = name.split(".", 1)[1]
        raise ValueError(
            f"Pass the short object-id (e.g. {hint!r}), "
            f"not the fully-qualified entity id {name!r}"
        )
    return f"{domain}.{name}"

in_domain

in_domain(domain: str) -> list[Entity]

Return all entities whose id starts with {domain}..

Source code in src/haclient/core/registry.py
101
102
103
104
def in_domain(self, domain: str) -> list[Entity]:
    """Return all entities whose id starts with ``{domain}.``."""
    prefix = f"{domain}."
    return [e for eid, e in self._entities.items() if eid.startswith(prefix)]