Index

Bus

Bases: Resource

Individual event bus instance for a specific owner (e.g., App or Service).

Source code in src/hassette/bus/bus.py

class Bus(Resource):
    """Individual event bus instance for a specific owner (e.g., App or Service)."""

    bus_service: "BusService"

    sync: BusSyncFacade
    """Synchronous facade for registering listeners from sync code (e.g. ``AppSync`` hooks)."""

    priority: int = 0
    """Priority level for event handlers created by this bus."""

    def __init__(self, hassette: "Hassette", *, priority: int = 0, parent: Resource | None = None) -> None:
        super().__init__(hassette, parent=parent)
        assert self.parent is not None, "Bus requires a parent Resource for telemetry identity (app_key/source_tier)"
        assert self.hassette._bus_service is not None, "Bus service not initialized"
        self.bus_service = self.hassette._bus_service
        self.priority = priority
        self._registered_handler_names: dict[tuple[str, int, str, str], str] = {}
        self._error_handler: BusErrorHandlerType | None = None
        self.sync = self.add_child(BusSyncFacade, bus=self)

    async def on_initialize(self) -> None:
        # Clear before any on() calls so partial-init failures don't leave stale keys.
        self._registered_handler_names.clear()
        self._error_handler = None
        self.mark_ready(reason="Bus initialized")

    async def on_shutdown(self) -> None:
        """Cleanup all listeners owned by this bus's owner on shutdown."""
        self.remove_all_listeners()

    def on_error(self, handler: "BusErrorHandlerType") -> None:
        """Register an app-level error handler for this bus.

        The handler is called when any listener on this bus raises an exception
        (including ``TimeoutError``) and the listener does not have its own
        per-registration error handler.

        This is an app-level fallback — it is resolved at dispatch time, not at listener
        registration time. A later call to ``on_error()`` replaces any previously registered
        handler.

        Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near
        app shutdown may be cancelled before they complete. Do not rely on error handlers
        for delivery-critical alerting during system teardown.

        Args:
            handler: A sync or async callable that accepts a :class:`~hassette.bus.error_context.BusErrorContext`.
        """
        self._error_handler = handler

    @property
    def config_log_level(self) -> LOG_LEVEL_TYPE:
        """Return the log level from the config for this resource."""
        return self.hassette.config.logging.bus_service

    async def add_listener(self, listener: "Listener") -> None:
        """Add a pre-built listener to the bus.

        This is the direct entry point for callers that construct a ``Listener``
        externally. The normal registration flow (``on_state_change``, ``on()``,
        etc.) goes through ``_on_internal`` instead.

        Raises:
            ListenerNameRequiredError: If the listener has no ``name`` (required for all DB-registered
                listeners, including once-listeners; cancel-listeners bypass this path entirely).
            DuplicateListenerError: If the listener's natural key is already registered on this bus instance.
        """
        if listener.identity.name is None:
            raise ListenerNameRequiredError(handler_method=listener.identity.handler_name, topic=listener.topic)
        self.register_and_check_collision(listener)
        await self.bus_service.add_listener(listener)

    def register_and_check_collision(self, listener: "Listener") -> None:
        """Register a non-once listener's natural key, raising DuplicateListenerError on a same-session clash.

        Mutates the per-bus key registry as a side effect — this is the single point where in-session
        duplicate detection happens. Once-listeners are exempt and are not registered.
        """
        if listener.options.once:
            return
        natural_key = self._listener_natural_key(listener)
        if natural_key in self._registered_handler_names:
            existing_handler = self._registered_handler_names[natural_key]
            raise DuplicateListenerError(
                name=listener.identity.name or "",
                topic=listener.topic,
                existing_handler=existing_handler,
                duplicate_handler=listener.identity.handler_name,
            )
        self._registered_handler_names[natural_key] = listener.identity.handler_name

    def _listener_natural_key(self, listener: "Listener") -> tuple[str, int, str, str]:
        """Compute the natural key tuple for a listener (for collision tracking).

        CANONICAL form: ``(app_key, instance_index, name, topic)``.
        Matches the SQL unique index defined in 001.sql and the repository upsert ON CONFLICT target.
        """
        return (
            listener.identity.app_key,
            listener.identity.instance_index,
            listener.identity.name or "",
            listener.topic,
        )

    def remove_listener(self, listener: "Listener") -> None:
        """Remove a listener from the bus."""
        natural_key = self._listener_natural_key(listener)
        self._registered_handler_names.pop(natural_key, None)
        self.bus_service.remove_listener(listener)

    def remove_all_listeners(self) -> None:
        """Remove all listeners owned by this bus's owner."""
        self._registered_handler_names.clear()
        self.bus_service.remove_listeners_by_owner(self.owner_id)

    def get_listeners(self) -> list["Listener"]:
        """Get all listeners owned by this bus's owner."""
        return self.bus_service.get_listeners_by_owner(self.owner_id)

    async def emit(self, topic: str, data: object) -> None:
        """Broadcast data to all subscribers of the given topic.

        Subscribers annotated with ``D.EventData[T]`` receive ``data`` pre-extracted.
        If the internal event stream is closed (during shutdown), the event is silently dropped.
        """
        payload = HassettePayload(data=data)
        event = Event(topic=topic, payload=payload)
        await self.hassette.send_event(event)

    async def on(
        self,
        *,
        topic: str,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        once: bool = False,
        debounce: float | None = None,
        throttle: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
    ) -> Subscription:
        """Subscribe to an event topic with optional filtering and modifiers.

        This is the public registration method for raw topic subscriptions. This method is
        ``async`` and must be awaited. Registration completes before the call returns —
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            topic: The event topic to listen to.
            handler: The function to call when the event matches.
            where: Optional predicates to filter events. These can be custom callables or predefined predicates from
                `hassette.event_handling.predicates`. They will receive the full event for evaluation.
            kwargs: Keyword arguments to pass to the handler.
            once: If True, the handler will be called only once and then removed.
            debounce: If set, applies a debounce to the handler.
            throttle: If set, applies a throttle to the handler.
            timeout: Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config.
                None means fall through to the config default.
            timeout_disabled: When True, disables timeout enforcement for this listener regardless of config.
            name: Required. Stable string identifier for this listener. Forms part of the natural
                key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
                restarts. Omitting it raises ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.

        Returns:
            A subscription object. ``sub.cancel()`` removes the listener.
            ``sub.listener.db_id`` is a valid integer immediately on return.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already registered.
        """
        return await self._on_internal(
            topic=topic,
            handler=handler,
            where=where,
            kwargs=kwargs,
            once=once,
            debounce=debounce,
            throttle=throttle,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            name=name,
            on_error=on_error,
            duration_config=None,
        )

    async def _on_internal(
        self,
        *,
        topic: str,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        once: bool = False,
        debounce: float | None = None,
        throttle: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        duration_config: "DurationConfig | None" = None,
    ) -> Subscription:
        """Private registration method carrying the full parameter set.

        Called by on() (with duration_config=None) and by _subscribe() (which
        builds DurationConfig from duration/entity_id when provided).

        Builds all sub-structs (ListenerIdentity, ListenerOptions, HandlerInvoker)
        here and calls Listener.create() via the sub-struct path. Source location is
        captured before construction (while user code is still on the stack).

        DB registration is awaited inline — the listener's db_id is set and the
        listener is routable before this method returns.
        """
        parent = self.parent
        assert parent is not None
        app_key = parent.app_key
        instance_index = parent.index
        source_tier = parent.source_tier
        assert source_tier in ("app", "framework"), f"Invalid source_tier={source_tier!r} on {parent.class_name}"

        # Capture source while user code is still on the stack (before async boundary)
        source_location, registration_source = capture_registration_source()

        handler_name = callable_name(handler)
        short_name = callable_short_name(handler)

        if name is None:
            raise ListenerNameRequiredError(handler_method=handler_name, topic=topic)

        identity = ListenerIdentity(
            owner_id=self.owner_id,
            app_key=app_key,
            instance_index=instance_index,
            name=name,
            source_tier=source_tier,
            handler_name=handler_name,
            handler_short_name=short_name,
            source_location=source_location,
            registration_source=registration_source or "",
        )

        options = ListenerOptions(
            once=once,
            debounce=debounce,
            throttle=throttle,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            priority=self.priority,
        )

        invoker = HandlerInvoker.create(
            task_bucket=self.task_bucket,
            handler=handler,
            kwargs=kwargs,
            options=options,
            error_handler=on_error,
            app_error_handler_resolver=lambda: self._error_handler,
        )

        listener = Listener.create(
            topic=topic,
            identity=identity,
            options=options,
            invoker=invoker,
            where=where,
            duration_config=duration_config,
            logger=self.logger,
        )

        self.register_and_check_collision(listener)

        def unsubscribe() -> None:
            self.remove_listener(listener)

        await self.bus_service.add_listener(listener)
        return Subscription(listener, unsubscribe)

    async def _subscribe(
        self,
        *,
        method_name: str,
        topic: str,
        handler: "HandlerType",
        preds: list["Predicate"],
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        log_params: Mapping[str, Any] | None = None,
        immediate: bool = False,
        duration: float | None = None,
        entity_id: str | None = None,
        is_attribute_listener: bool = False,
        hold_preds: list["Predicate"] | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Common subscription tail: log, normalize where, delegate to _on_internal()."""
        if self.logger.isEnabledFor(logging.DEBUG):
            filtered = (
                {k: v for k, v in log_params.items() if v is not None and not isinstance(v, Sentinel)}
                if log_params
                else {}
            )
            params_str = ", ".join(f"{k}='{v}'" for k, v in filtered.items())

            self.logger.debug(
                "Subscribing to %s with %s - being handled by '%s'",
                method_name,
                params_str,
                callable_short_name(handler),
            )

        if where is not None:
            normalized_where = where if callable(where) else P.AllOf.ensure_iterable(where)
            preds.append(normalized_where)
            if hold_preds is not None:
                hold_preds = [*hold_preds, normalized_where]

        # Build DurationConfig when entity_id is provided (for duration or immediate listeners)
        duration_config: DurationConfig | None = None
        if entity_id:
            duration_config = DurationConfig(
                entity_id=entity_id,
                duration=duration,
                immediate=immediate,
                is_attribute_listener=is_attribute_listener,
                hold_predicate=P.AllOf.ensure_iterable(hold_preds) if hold_preds else None,
            )

        return await self._on_internal(
            topic=topic,
            handler=handler,
            where=preds,
            kwargs=kwargs,
            duration_config=duration_config,
            name=name,
            on_error=on_error,
            **opts,
        )

    @staticmethod
    def _normalize_service_where(
        preds: list["Predicate"],
        where: "Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None",
    ) -> None:
        """Normalize on_call_service's Mapping-aware where clause into predicates."""
        if where is None:
            return

        if isinstance(where, Mapping):
            preds.append(P.ServiceDataWhere(where))
        elif callable(where):
            preds.append(where)
        else:
            mappings = [w for w in where if isinstance(w, Mapping)]
            other = [w for w in where if not isinstance(w, Mapping)]

            preds.extend(P.ServiceDataWhere(w) for w in mappings)

            if other:
                preds.append(P.AllOf.ensure_iterable(other))

    async def on_state_change(
        self,
        entity_id: str,
        *,
        handler: "HandlerType",
        changed: bool | ComparisonCondition = True,
        changed_from: "ChangeType" = NOT_PROVIDED,
        changed_to: "ChangeType" = NOT_PROVIDED,
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        immediate: bool = False,
        duration: float | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to state changes for a specific entity.

        This method is ``async`` and must be awaited. Registration — routing and database
        persistence — completes before the call returns. ``sub.listener.db_id`` is a valid
        integer immediately on return.

        Args:
            entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
            handler: The function to call when the event matches.
            changed: Whether to filter only events where the state changed. If a ComparisonCondition is provided, it
                will be used to compare the old and new state values.
            changed_from: A value or callable that will be used to filter state changes *from* this value.
            changed_to: A value or callable that will be used to filter state changes *to* this value.
            where: Additional predicates to filter events (e.g. ValueIs) or custom callables. These will receive the
                full event for evaluation.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. A stable string identifier for this listener. Forms part of the natural
                key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
                restarts. Omitting it raises ``ListenerNameRequiredError`` at call time.
            **opts: Additional options like `once`, `debounce` and `throttle`.

        Returns:
            A subscription object. ``sub.listener.db_id`` is set immediately. ``sub.cancel()``
            removes the listener from routing.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
                registered on this bus in the current session.
        """
        if immediate and is_glob(entity_id):
            raise ValueError(
                f"'immediate=True' is not supported with glob patterns. "
                f"entity_id={entity_id!r} contains glob characters."
            )
        if duration is not None and is_glob(entity_id):
            raise ValueError(
                f"'duration' is not supported with glob patterns. entity_id={entity_id!r} contains glob characters."
            )

        preds, hold_preds = build_state_preds(
            entity_id, changed=changed, changed_from=changed_from, changed_to=changed_to
        )

        return await self._subscribe(
            method_name=f"entity '{entity_id}'",
            topic=f"{Topic.HASS_EVENT_STATE_CHANGED!s}.{entity_id}",
            handler=handler,
            preds=preds,
            where=where,
            kwargs=kwargs,
            log_params={"changed": changed, "changed_from": changed_from, "changed_to": changed_to, "where": where},
            immediate=immediate,
            duration=duration,
            entity_id=entity_id,
            hold_preds=hold_preds if duration is not None else None,
            name=name,
            on_error=on_error,
            **opts,
        )

    async def on_attribute_change(
        self,
        entity_id: str,
        attr: str,
        *,
        handler: "HandlerType",
        changed: bool | ComparisonCondition = True,
        changed_from: "ChangeType" = NOT_PROVIDED,
        changed_to: "ChangeType" = NOT_PROVIDED,
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        immediate: bool = False,
        duration: float | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to state change events for a specific entity's attribute.

        This method is ``async`` and must be awaited. Registration completes before the call
        returns. ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
            attr: The attribute name to filter changes on (e.g., "volume").
            handler: The function to call when the event matches.
            changed: Whether to filter only events where the attribute changed. If a ComparisonCondition is provided,
                it will be used to compare the old and new attribute values.
            changed_from: A value or callable that will be used to filter attribute changes *from* this value.
            changed_to: A value or callable that will be used to filter attribute changes *to* this value.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable string identifier. Omitting it raises ``ListenerNameRequiredError``.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object. ``sub.listener.db_id`` is set immediately.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already registered.
        """
        if immediate and is_glob(entity_id):
            raise ValueError(
                f"'immediate=True' is not supported with glob patterns. "
                f"entity_id={entity_id!r} contains glob characters."
            )
        if duration is not None and is_glob(entity_id):
            raise ValueError(
                f"'duration' is not supported with glob patterns. entity_id={entity_id!r} contains glob characters."
            )

        if not changed:
            self.logger.warning(
                (
                    "Handler '%s' - attribute change subscription "
                    "will fire on every change event for '%s' due to 'changed=False'. "
                    "Consider using `on_state_change` with 'changed=False' instead for clarity."
                ),
                callable_short_name(handler),
                entity_id,
            )

        preds, hold_preds = build_attr_preds(
            entity_id, attr, changed=changed, changed_from=changed_from, changed_to=changed_to
        )

        return await self._subscribe(
            method_name=f"entity '{entity_id}' attribute '{attr}'",
            topic=f"{Topic.HASS_EVENT_STATE_CHANGED!s}.{entity_id}",
            handler=handler,
            preds=preds,
            where=where,
            kwargs=kwargs,
            log_params={"changed_from": changed_from, "changed_to": changed_to, "where": where},
            immediate=immediate,
            duration=duration,
            entity_id=entity_id,
            hold_preds=hold_preds if duration is not None else None,
            is_attribute_listener=True,
            name=name,
            on_error=on_error,
            **opts,
        )

    async def on_call_service(
        self,
        domain: str | None = None,
        service: str | None = None,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to service call events.

        This method is ``async`` and must be awaited. Registration completes before the call
        returns. ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            domain: The domain to filter service calls (e.g., "light").
            service: The service to filter service calls (e.g., "turn_on").
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable string identifier for this listener. Omitting it raises
                ``ListenerNameRequiredError`` at call time.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object. ``sub.listener.db_id`` is set immediately.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already registered.
        """

        preds: list[Predicate] = []
        if domain is not None:
            preds.append(P.DomainMatches(domain))

        if service is not None:
            preds.append(P.ServiceMatches(service))

        self._normalize_service_where(preds, where)

        return await self._subscribe(
            method_name="call_service",
            topic=Topic.HASS_EVENT_CALL_SERVICE,
            handler=handler,
            preds=preds,
            where=None,
            kwargs=kwargs,
            log_params={"domain": domain, "service": service, "where": where},
            name=name,
            on_error=on_error,
            **opts,
        )

    async def on_component_loaded(
        self,
        component: str | None = None,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to component loaded events.

        Args:
            component: The component to filter load events (e.g., "light").
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener.
        """

        preds: list[Predicate] = []

        if component is not None:
            preds.append(P.ValueIs(source=get_path("payload.data.component"), condition=component))

        return await self._subscribe(
            method_name="component_loaded",
            topic=Topic.HASS_EVENT_COMPONENT_LOADED,
            handler=handler,
            preds=preds,
            where=where,
            kwargs=kwargs,
            log_params={"component": component, "where": where},
            name=name,
            on_error=on_error,
            **opts,
        )

    async def on_service_registered(
        self,
        domain: str | None = None,
        service: str | None = None,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to service registered events.

        Args:
            domain: The domain to filter service registrations (e.g., "light").
            service: The service to filter service registrations (e.g., "turn_on").
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener.
        """

        preds: list[Predicate] = []

        if domain is not None:
            preds.append(P.DomainMatches(domain))

        if service is not None:
            preds.append(P.ServiceMatches(service))

        return await self._subscribe(
            method_name="service_registered",
            topic=Topic.HASS_EVENT_SERVICE_REGISTERED,
            handler=handler,
            preds=preds,
            where=where,
            kwargs=kwargs,
            log_params={"domain": domain, "service": service, "where": where},
            name=name,
            on_error=on_error,
            **opts,
        )

    async def on_homeassistant_restart(
        self,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to Home Assistant restart events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return await self.on_call_service(
            domain="homeassistant", service="restart", handler=handler, where=where, kwargs=kwargs, name=name, **opts
        )

    async def on_homeassistant_start(
        self,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to Home Assistant start events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return await self.on_call_service(
            domain="homeassistant", service="start", handler=handler, where=where, kwargs=kwargs, name=name, **opts
        )

    async def on_homeassistant_stop(
        self,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to Home Assistant stop events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener.
        """
        return await self.on_call_service(
            domain="homeassistant", service="stop", handler=handler, where=where, kwargs=kwargs, name=name, **opts
        )

    async def on_hassette_service_status(
        self,
        status: ResourceStatus | None = None,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to hassette service status events.

        Args:
            status: The status to filter events (e.g., ResourceStatus.STARTED).
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. A stable string identifier for this listener. Forms part of the
                natural key ``(app_key, instance_index, name, topic)`` used for upsert
                deduplication across restarts. Omitting it raises ``ListenerNameRequiredError``
                at call time.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener.
        """

        preds: list[Predicate] = []

        if status is not None:
            preds.append(P.ValueIs(source=get_path("payload.data.status"), condition=status))

        return await self._subscribe(
            method_name="hassette.service_status",
            topic=Topic.HASSETTE_EVENT_SERVICE_STATUS,
            handler=handler,
            preds=preds,
            where=where,
            kwargs=kwargs,
            log_params={"status": status, "where": where},
            name=name,
            on_error=on_error,
            **opts,
        )

    async def on_hassette_service_failed(
        self,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to hassette service failed events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. A stable string identifier for this listener. Forms part of the
                natural key ``(app_key, instance_index, name, topic)`` used for upsert
                deduplication across restarts. Omitting it raises ``ListenerNameRequiredError``
                at call time.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener.
        """

        return await self.on_hassette_service_status(
            status=ResourceStatus.FAILED, handler=handler, where=where, kwargs=kwargs, name=name, **opts
        )

    async def on_hassette_service_crashed(
        self,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to hassette service crashed events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener.
        """

        return await self.on_hassette_service_status(
            status=ResourceStatus.CRASHED, handler=handler, where=where, kwargs=kwargs, name=name, **opts
        )

    async def on_hassette_service_started(
        self,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to hassette service started events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener.
        """

        return await self.on_hassette_service_status(
            status=ResourceStatus.RUNNING, handler=handler, where=where, kwargs=kwargs, name=name, **opts
        )

    async def on_websocket_connected(
        self,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to websocket connected events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener.
        """

        return await self.on(
            topic=Topic.HASSETTE_EVENT_WEBSOCKET_CONNECTED,
            handler=handler,
            where=where,
            kwargs=kwargs,
            name=name,
            **opts,
        )

    async def on_websocket_disconnected(
        self,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to websocket disconnected events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener.
        """

        return await self.on(
            topic=Topic.HASSETTE_EVENT_WEBSOCKET_DISCONNECTED,
            handler=handler,
            where=where,
            kwargs=kwargs,
            name=name,
            **opts,
        )

    async def on_app_state_changed(
        self,
        *,
        handler: "HandlerType",
        app_key: str | None = None,
        status: ResourceStatus | None = None,
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to app instance state change events.

        Args:
            handler: The function to call when the event matches.
            app_key: Filter events for a specific app key.
            status: Filter events for a specific status.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener.
        """

        preds: list[Predicate] = []

        if app_key is not None:
            preds.append(P.ValueIs(source=get_path("payload.data.app_key"), condition=app_key))

        if status is not None:
            preds.append(P.ValueIs(source=get_path("payload.data.status"), condition=status))

        return await self._subscribe(
            method_name="app_state_changed",
            topic=Topic.HASSETTE_EVENT_APP_STATE_CHANGED,
            handler=handler,
            preds=preds,
            where=where,
            kwargs=kwargs,
            log_params={"app_key": app_key, "status": status, "where": where},
            name=name,
            on_error=on_error,
            **opts,
        )

    async def on_app_running(
        self,
        *,
        handler: "HandlerType",
        app_key: str | None = None,
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to app instances reaching RUNNING status.

        Args:
            handler: The function to call when the event matches.
            app_key: Filter events for a specific app key.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener.
        """

        return await self.on_app_state_changed(
            handler=handler,
            app_key=app_key,
            status=ResourceStatus.RUNNING,
            where=where,
            kwargs=kwargs,
            name=name,
            **opts,
        )

    async def on_app_stopping(
        self,
        *,
        handler: "HandlerType",
        app_key: str | None = None,
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to app instances entering STOPPING status.

        Args:
            handler: The function to call when the event matches.
            app_key: Filter events for a specific app key.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener.
        """

        return await self.on_app_state_changed(
            handler=handler,
            app_key=app_key,
            status=ResourceStatus.STOPPING,
            where=where,
            kwargs=kwargs,
            name=name,
            **opts,
        )

sync: BusSyncFacade = self.add_child(BusSyncFacade, bus=self) instance-attribute

Synchronous facade for registering listeners from sync code (e.g. AppSync hooks).

priority: int = priority class-attribute instance-attribute

Priority level for event handlers created by this bus.

config_log_level: LOG_LEVEL_TYPE property

Return the log level from the config for this resource.

on_shutdown() -> None async

Cleanup all listeners owned by this bus's owner on shutdown.

Source code in src/hassette/bus/bus.py

async def on_shutdown(self) -> None:
    """Cleanup all listeners owned by this bus's owner on shutdown."""
    self.remove_all_listeners()

on_error(handler: BusErrorHandlerType) -> None

Register an app-level error handler for this bus.

The handler is called when any listener on this bus raises an exception (including TimeoutError) and the listener does not have its own per-registration error handler.

This is an app-level fallback — it is resolved at dispatch time, not at listener registration time. A later call to on_error() replaces any previously registered handler.

Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near app shutdown may be cancelled before they complete. Do not rely on error handlers for delivery-critical alerting during system teardown.

Parameters:

Name	Type	Description	Default
handler	BusErrorHandlerType	A sync or async callable that accepts a :class:~hassette.bus.error_context.BusErrorContext.	required

Source code in src/hassette/bus/bus.py

def on_error(self, handler: "BusErrorHandlerType") -> None:
    """Register an app-level error handler for this bus.

    The handler is called when any listener on this bus raises an exception
    (including ``TimeoutError``) and the listener does not have its own
    per-registration error handler.

    This is an app-level fallback — it is resolved at dispatch time, not at listener
    registration time. A later call to ``on_error()`` replaces any previously registered
    handler.

    Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near
    app shutdown may be cancelled before they complete. Do not rely on error handlers
    for delivery-critical alerting during system teardown.

    Args:
        handler: A sync or async callable that accepts a :class:`~hassette.bus.error_context.BusErrorContext`.
    """
    self._error_handler = handler

add_listener(listener: Listener) -> None async

Add a pre-built listener to the bus.

This is the direct entry point for callers that construct a Listener externally. The normal registration flow (on_state_change, on(), etc.) goes through _on_internal instead.

Raises:

Type	Description
ListenerNameRequiredError	If the listener has no name (required for all DB-registered listeners, including once-listeners; cancel-listeners bypass this path entirely).
DuplicateListenerError	If the listener's natural key is already registered on this bus instance.

Source code in src/hassette/bus/bus.py

async def add_listener(self, listener: "Listener") -> None:
    """Add a pre-built listener to the bus.

    This is the direct entry point for callers that construct a ``Listener``
    externally. The normal registration flow (``on_state_change``, ``on()``,
    etc.) goes through ``_on_internal`` instead.

    Raises:
        ListenerNameRequiredError: If the listener has no ``name`` (required for all DB-registered
            listeners, including once-listeners; cancel-listeners bypass this path entirely).
        DuplicateListenerError: If the listener's natural key is already registered on this bus instance.
    """
    if listener.identity.name is None:
        raise ListenerNameRequiredError(handler_method=listener.identity.handler_name, topic=listener.topic)
    self.register_and_check_collision(listener)
    await self.bus_service.add_listener(listener)

register_and_check_collision(listener: Listener) -> None

Register a non-once listener's natural key, raising DuplicateListenerError on a same-session clash.

Mutates the per-bus key registry as a side effect — this is the single point where in-session duplicate detection happens. Once-listeners are exempt and are not registered.

Source code in src/hassette/bus/bus.py

def register_and_check_collision(self, listener: "Listener") -> None:
    """Register a non-once listener's natural key, raising DuplicateListenerError on a same-session clash.

    Mutates the per-bus key registry as a side effect — this is the single point where in-session
    duplicate detection happens. Once-listeners are exempt and are not registered.
    """
    if listener.options.once:
        return
    natural_key = self._listener_natural_key(listener)
    if natural_key in self._registered_handler_names:
        existing_handler = self._registered_handler_names[natural_key]
        raise DuplicateListenerError(
            name=listener.identity.name or "",
            topic=listener.topic,
            existing_handler=existing_handler,
            duplicate_handler=listener.identity.handler_name,
        )
    self._registered_handler_names[natural_key] = listener.identity.handler_name

remove_listener(listener: Listener) -> None

Remove a listener from the bus.

Source code in src/hassette/bus/bus.py

def remove_listener(self, listener: "Listener") -> None:
    """Remove a listener from the bus."""
    natural_key = self._listener_natural_key(listener)
    self._registered_handler_names.pop(natural_key, None)
    self.bus_service.remove_listener(listener)

remove_all_listeners() -> None

Remove all listeners owned by this bus's owner.

Source code in src/hassette/bus/bus.py

def remove_all_listeners(self) -> None:
    """Remove all listeners owned by this bus's owner."""
    self._registered_handler_names.clear()
    self.bus_service.remove_listeners_by_owner(self.owner_id)

get_listeners() -> list[Listener]

Get all listeners owned by this bus's owner.

Source code in src/hassette/bus/bus.py

def get_listeners(self) -> list["Listener"]:
    """Get all listeners owned by this bus's owner."""
    return self.bus_service.get_listeners_by_owner(self.owner_id)

emit(topic: str, data: object) -> None async

Broadcast data to all subscribers of the given topic.

Subscribers annotated with D.EventData[T] receive data pre-extracted. If the internal event stream is closed (during shutdown), the event is silently dropped.

Source code in src/hassette/bus/bus.py

async def emit(self, topic: str, data: object) -> None:
    """Broadcast data to all subscribers of the given topic.

    Subscribers annotated with ``D.EventData[T]`` receive ``data`` pre-extracted.
    If the internal event stream is closed (during shutdown), the event is silently dropped.
    """
    payload = HassettePayload(data=data)
    event = Event(topic=topic, payload=payload)
    await self.hassette.send_event(event)

on(*, topic: str, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, once: bool = False, debounce: float | None = None, throttle: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, name: str | None = None, on_error: BusErrorHandlerType | None = None) -> Subscription async

Subscribe to an event topic with optional filtering and modifiers.

This is the public registration method for raw topic subscriptions. This method is async and must be awaited. Registration completes before the call returns — sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
topic	str	The event topic to listen to.	required
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Optional predicates to filter events. These can be custom callables or predefined predicates from hassette.event_handling.predicates. They will receive the full event for evaluation.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
once	bool	If True, the handler will be called only once and then removed.	False
debounce	float | None	If set, applies a debounce to the handler.	None
throttle	float | None	If set, applies a throttle to the handler.	None
timeout	float | None	Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config. None means fall through to the config default.	None
timeout_disabled	bool	When True, disables timeout enforcement for this listener regardless of config.	False
name	str | None	Required. Stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it raises ListenerNameRequiredError at call time.	None
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None

Returns:

Type	Description
Subscription	A subscription object. sub.cancel() removes the listener.
Subscription	sub.listener.db_id is a valid integer immediately on return.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered.

Source code in src/hassette/bus/bus.py

async def on(
    self,
    *,
    topic: str,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    once: bool = False,
    debounce: float | None = None,
    throttle: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
) -> Subscription:
    """Subscribe to an event topic with optional filtering and modifiers.

    This is the public registration method for raw topic subscriptions. This method is
    ``async`` and must be awaited. Registration completes before the call returns —
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        topic: The event topic to listen to.
        handler: The function to call when the event matches.
        where: Optional predicates to filter events. These can be custom callables or predefined predicates from
            `hassette.event_handling.predicates`. They will receive the full event for evaluation.
        kwargs: Keyword arguments to pass to the handler.
        once: If True, the handler will be called only once and then removed.
        debounce: If set, applies a debounce to the handler.
        throttle: If set, applies a throttle to the handler.
        timeout: Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config.
            None means fall through to the config default.
        timeout_disabled: When True, disables timeout enforcement for this listener regardless of config.
        name: Required. Stable string identifier for this listener. Forms part of the natural
            key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
            restarts. Omitting it raises ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.

    Returns:
        A subscription object. ``sub.cancel()`` removes the listener.
        ``sub.listener.db_id`` is a valid integer immediately on return.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already registered.
    """
    return await self._on_internal(
        topic=topic,
        handler=handler,
        where=where,
        kwargs=kwargs,
        once=once,
        debounce=debounce,
        throttle=throttle,
        timeout=timeout,
        timeout_disabled=timeout_disabled,
        name=name,
        on_error=on_error,
        duration_config=None,
    )

on_state_change(entity_id: str, *, handler: HandlerType, changed: bool | ComparisonCondition = True, changed_from: ChangeType = NOT_PROVIDED, changed_to: ChangeType = NOT_PROVIDED, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, immediate: bool = False, duration: float | None = None, name: str | None = None, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to state changes for a specific entity.

This method is async and must be awaited. Registration — routing and database persistence — completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
entity_id	str	The entity ID to filter events for (e.g., "media_player.living_room_speaker").	required
handler	HandlerType	The function to call when the event matches.	required
changed	bool | ComparisonCondition	Whether to filter only events where the state changed. If a ComparisonCondition is provided, it will be used to compare the old and new state values.	True
changed_from	ChangeType	A value or callable that will be used to filter state changes from this value.	NOT_PROVIDED
changed_to	ChangeType	A value or callable that will be used to filter state changes to this value.	NOT_PROVIDED
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events (e.g. ValueIs) or custom callables. These will receive the full event for evaluation.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Required. A stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it raises ListenerNameRequiredError at call time.	None
**opts	Unpack[Options]	Additional options like once, debounce and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object. sub.listener.db_id is set immediately. sub.cancel()
Subscription	removes the listener from routing.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered on this bus in the current session.

Source code in src/hassette/bus/bus.py

async def on_state_change(
    self,
    entity_id: str,
    *,
    handler: "HandlerType",
    changed: bool | ComparisonCondition = True,
    changed_from: "ChangeType" = NOT_PROVIDED,
    changed_to: "ChangeType" = NOT_PROVIDED,
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    immediate: bool = False,
    duration: float | None = None,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to state changes for a specific entity.

    This method is ``async`` and must be awaited. Registration — routing and database
    persistence — completes before the call returns. ``sub.listener.db_id`` is a valid
    integer immediately on return.

    Args:
        entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
        handler: The function to call when the event matches.
        changed: Whether to filter only events where the state changed. If a ComparisonCondition is provided, it
            will be used to compare the old and new state values.
        changed_from: A value or callable that will be used to filter state changes *from* this value.
        changed_to: A value or callable that will be used to filter state changes *to* this value.
        where: Additional predicates to filter events (e.g. ValueIs) or custom callables. These will receive the
            full event for evaluation.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. A stable string identifier for this listener. Forms part of the natural
            key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
            restarts. Omitting it raises ``ListenerNameRequiredError`` at call time.
        **opts: Additional options like `once`, `debounce` and `throttle`.

    Returns:
        A subscription object. ``sub.listener.db_id`` is set immediately. ``sub.cancel()``
        removes the listener from routing.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
            registered on this bus in the current session.
    """
    if immediate and is_glob(entity_id):
        raise ValueError(
            f"'immediate=True' is not supported with glob patterns. "
            f"entity_id={entity_id!r} contains glob characters."
        )
    if duration is not None and is_glob(entity_id):
        raise ValueError(
            f"'duration' is not supported with glob patterns. entity_id={entity_id!r} contains glob characters."
        )

    preds, hold_preds = build_state_preds(
        entity_id, changed=changed, changed_from=changed_from, changed_to=changed_to
    )

    return await self._subscribe(
        method_name=f"entity '{entity_id}'",
        topic=f"{Topic.HASS_EVENT_STATE_CHANGED!s}.{entity_id}",
        handler=handler,
        preds=preds,
        where=where,
        kwargs=kwargs,
        log_params={"changed": changed, "changed_from": changed_from, "changed_to": changed_to, "where": where},
        immediate=immediate,
        duration=duration,
        entity_id=entity_id,
        hold_preds=hold_preds if duration is not None else None,
        name=name,
        on_error=on_error,
        **opts,
    )

on_attribute_change(entity_id: str, attr: str, *, handler: HandlerType, changed: bool | ComparisonCondition = True, changed_from: ChangeType = NOT_PROVIDED, changed_to: ChangeType = NOT_PROVIDED, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, immediate: bool = False, duration: float | None = None, name: str | None = None, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to state change events for a specific entity's attribute.

This method is async and must be awaited. Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
entity_id	str	The entity ID to filter events for (e.g., "media_player.living_room_speaker").	required
attr	str	The attribute name to filter changes on (e.g., "volume").	required
handler	HandlerType	The function to call when the event matches.	required
changed	bool | ComparisonCondition	Whether to filter only events where the attribute changed. If a ComparisonCondition is provided, it will be used to compare the old and new attribute values.	True
changed_from	ChangeType	A value or callable that will be used to filter attribute changes from this value.	NOT_PROVIDED
changed_to	ChangeType	A value or callable that will be used to filter attribute changes to this value.	NOT_PROVIDED
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Required. Stable string identifier. Omitting it raises ListenerNameRequiredError.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object. sub.listener.db_id is set immediately.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered.

Source code in src/hassette/bus/bus.py

async def on_attribute_change(
    self,
    entity_id: str,
    attr: str,
    *,
    handler: "HandlerType",
    changed: bool | ComparisonCondition = True,
    changed_from: "ChangeType" = NOT_PROVIDED,
    changed_to: "ChangeType" = NOT_PROVIDED,
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    immediate: bool = False,
    duration: float | None = None,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to state change events for a specific entity's attribute.

    This method is ``async`` and must be awaited. Registration completes before the call
    returns. ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
        attr: The attribute name to filter changes on (e.g., "volume").
        handler: The function to call when the event matches.
        changed: Whether to filter only events where the attribute changed. If a ComparisonCondition is provided,
            it will be used to compare the old and new attribute values.
        changed_from: A value or callable that will be used to filter attribute changes *from* this value.
        changed_to: A value or callable that will be used to filter attribute changes *to* this value.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable string identifier. Omitting it raises ``ListenerNameRequiredError``.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object. ``sub.listener.db_id`` is set immediately.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already registered.
    """
    if immediate and is_glob(entity_id):
        raise ValueError(
            f"'immediate=True' is not supported with glob patterns. "
            f"entity_id={entity_id!r} contains glob characters."
        )
    if duration is not None and is_glob(entity_id):
        raise ValueError(
            f"'duration' is not supported with glob patterns. entity_id={entity_id!r} contains glob characters."
        )

    if not changed:
        self.logger.warning(
            (
                "Handler '%s' - attribute change subscription "
                "will fire on every change event for '%s' due to 'changed=False'. "
                "Consider using `on_state_change` with 'changed=False' instead for clarity."
            ),
            callable_short_name(handler),
            entity_id,
        )

    preds, hold_preds = build_attr_preds(
        entity_id, attr, changed=changed, changed_from=changed_from, changed_to=changed_to
    )

    return await self._subscribe(
        method_name=f"entity '{entity_id}' attribute '{attr}'",
        topic=f"{Topic.HASS_EVENT_STATE_CHANGED!s}.{entity_id}",
        handler=handler,
        preds=preds,
        where=where,
        kwargs=kwargs,
        log_params={"changed_from": changed_from, "changed_to": changed_to, "where": where},
        immediate=immediate,
        duration=duration,
        entity_id=entity_id,
        hold_preds=hold_preds if duration is not None else None,
        is_attribute_listener=True,
        name=name,
        on_error=on_error,
        **opts,
    )

on_call_service(domain: str | None = None, service: str | None = None, *, handler: HandlerType, where: Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to service call events.

This method is async and must be awaited. Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
domain	str | None	The domain to filter service calls (e.g., "light").	None
service	str | None	The service to filter service calls (e.g., "turn_on").	None
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Required. Stable string identifier for this listener. Omitting it raises ListenerNameRequiredError at call time.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object. sub.listener.db_id is set immediately.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered.

Source code in src/hassette/bus/bus.py

async def on_call_service(
    self,
    domain: str | None = None,
    service: str | None = None,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to service call events.

    This method is ``async`` and must be awaited. Registration completes before the call
    returns. ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        domain: The domain to filter service calls (e.g., "light").
        service: The service to filter service calls (e.g., "turn_on").
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable string identifier for this listener. Omitting it raises
            ``ListenerNameRequiredError`` at call time.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object. ``sub.listener.db_id`` is set immediately.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already registered.
    """

    preds: list[Predicate] = []
    if domain is not None:
        preds.append(P.DomainMatches(domain))

    if service is not None:
        preds.append(P.ServiceMatches(service))

    self._normalize_service_where(preds, where)

    return await self._subscribe(
        method_name="call_service",
        topic=Topic.HASS_EVENT_CALL_SERVICE,
        handler=handler,
        preds=preds,
        where=None,
        kwargs=kwargs,
        log_params={"domain": domain, "service": service, "where": where},
        name=name,
        on_error=on_error,
        **opts,
    )

on_component_loaded(component: str | None = None, *, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to component loaded events.

Parameters:

Name	Type	Description	Default
component	str | None	The component to filter load events (e.g., "light").	None
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

async def on_component_loaded(
    self,
    component: str | None = None,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to component loaded events.

    Args:
        component: The component to filter load events (e.g., "light").
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener.
    """

    preds: list[Predicate] = []

    if component is not None:
        preds.append(P.ValueIs(source=get_path("payload.data.component"), condition=component))

    return await self._subscribe(
        method_name="component_loaded",
        topic=Topic.HASS_EVENT_COMPONENT_LOADED,
        handler=handler,
        preds=preds,
        where=where,
        kwargs=kwargs,
        log_params={"component": component, "where": where},
        name=name,
        on_error=on_error,
        **opts,
    )

on_service_registered(domain: str | None = None, service: str | None = None, *, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to service registered events.

Parameters:

Name	Type	Description	Default
domain	str | None	The domain to filter service registrations (e.g., "light").	None
service	str | None	The service to filter service registrations (e.g., "turn_on").	None
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

async def on_service_registered(
    self,
    domain: str | None = None,
    service: str | None = None,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to service registered events.

    Args:
        domain: The domain to filter service registrations (e.g., "light").
        service: The service to filter service registrations (e.g., "turn_on").
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener.
    """

    preds: list[Predicate] = []

    if domain is not None:
        preds.append(P.DomainMatches(domain))

    if service is not None:
        preds.append(P.ServiceMatches(service))

    return await self._subscribe(
        method_name="service_registered",
        topic=Topic.HASS_EVENT_SERVICE_REGISTERED,
        handler=handler,
        preds=preds,
        where=where,
        kwargs=kwargs,
        log_params={"domain": domain, "service": service, "where": where},
        name=name,
        on_error=on_error,
        **opts,
    )

on_homeassistant_restart(handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to Home Assistant restart events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

async def on_homeassistant_restart(
    self,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to Home Assistant restart events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return await self.on_call_service(
        domain="homeassistant", service="restart", handler=handler, where=where, kwargs=kwargs, name=name, **opts
    )

on_homeassistant_start(handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to Home Assistant start events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

async def on_homeassistant_start(
    self,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to Home Assistant start events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return await self.on_call_service(
        domain="homeassistant", service="start", handler=handler, where=where, kwargs=kwargs, name=name, **opts
    )

on_homeassistant_stop(handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to Home Assistant stop events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

async def on_homeassistant_stop(
    self,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to Home Assistant stop events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener.
    """
    return await self.on_call_service(
        domain="homeassistant", service="stop", handler=handler, where=where, kwargs=kwargs, name=name, **opts
    )

on_hassette_service_status(status: ResourceStatus | None = None, *, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to hassette service status events.

Parameters:

Name	Type	Description	Default
status	ResourceStatus | None	The status to filter events (e.g., ResourceStatus.STARTED).	None
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Required. A stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it raises ListenerNameRequiredError at call time.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

async def on_hassette_service_status(
    self,
    status: ResourceStatus | None = None,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to hassette service status events.

    Args:
        status: The status to filter events (e.g., ResourceStatus.STARTED).
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. A stable string identifier for this listener. Forms part of the
            natural key ``(app_key, instance_index, name, topic)`` used for upsert
            deduplication across restarts. Omitting it raises ``ListenerNameRequiredError``
            at call time.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener.
    """

    preds: list[Predicate] = []

    if status is not None:
        preds.append(P.ValueIs(source=get_path("payload.data.status"), condition=status))

    return await self._subscribe(
        method_name="hassette.service_status",
        topic=Topic.HASSETTE_EVENT_SERVICE_STATUS,
        handler=handler,
        preds=preds,
        where=where,
        kwargs=kwargs,
        log_params={"status": status, "where": where},
        name=name,
        on_error=on_error,
        **opts,
    )

on_hassette_service_failed(*, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to hassette service failed events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Required. A stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it raises ListenerNameRequiredError at call time.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

async def on_hassette_service_failed(
    self,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to hassette service failed events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. A stable string identifier for this listener. Forms part of the
            natural key ``(app_key, instance_index, name, topic)`` used for upsert
            deduplication across restarts. Omitting it raises ``ListenerNameRequiredError``
            at call time.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener.
    """

    return await self.on_hassette_service_status(
        status=ResourceStatus.FAILED, handler=handler, where=where, kwargs=kwargs, name=name, **opts
    )

on_hassette_service_crashed(*, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to hassette service crashed events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

async def on_hassette_service_crashed(
    self,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to hassette service crashed events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener.
    """

    return await self.on_hassette_service_status(
        status=ResourceStatus.CRASHED, handler=handler, where=where, kwargs=kwargs, name=name, **opts
    )

on_hassette_service_started(*, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to hassette service started events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

async def on_hassette_service_started(
    self,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to hassette service started events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener.
    """

    return await self.on_hassette_service_status(
        status=ResourceStatus.RUNNING, handler=handler, where=where, kwargs=kwargs, name=name, **opts
    )

on_websocket_connected(*, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to websocket connected events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

async def on_websocket_connected(
    self,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to websocket connected events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener.
    """

    return await self.on(
        topic=Topic.HASSETTE_EVENT_WEBSOCKET_CONNECTED,
        handler=handler,
        where=where,
        kwargs=kwargs,
        name=name,
        **opts,
    )

on_websocket_disconnected(*, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to websocket disconnected events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

async def on_websocket_disconnected(
    self,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to websocket disconnected events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener.
    """

    return await self.on(
        topic=Topic.HASSETTE_EVENT_WEBSOCKET_DISCONNECTED,
        handler=handler,
        where=where,
        kwargs=kwargs,
        name=name,
        **opts,
    )

on_app_state_changed(*, handler: HandlerType, app_key: str | None = None, status: ResourceStatus | None = None, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to app instance state change events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
app_key	str | None	Filter events for a specific app key.	None
status	ResourceStatus | None	Filter events for a specific status.	None
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

async def on_app_state_changed(
    self,
    *,
    handler: "HandlerType",
    app_key: str | None = None,
    status: ResourceStatus | None = None,
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to app instance state change events.

    Args:
        handler: The function to call when the event matches.
        app_key: Filter events for a specific app key.
        status: Filter events for a specific status.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener.
    """

    preds: list[Predicate] = []

    if app_key is not None:
        preds.append(P.ValueIs(source=get_path("payload.data.app_key"), condition=app_key))

    if status is not None:
        preds.append(P.ValueIs(source=get_path("payload.data.status"), condition=status))

    return await self._subscribe(
        method_name="app_state_changed",
        topic=Topic.HASSETTE_EVENT_APP_STATE_CHANGED,
        handler=handler,
        preds=preds,
        where=where,
        kwargs=kwargs,
        log_params={"app_key": app_key, "status": status, "where": where},
        name=name,
        on_error=on_error,
        **opts,
    )

on_app_running(*, handler: HandlerType, app_key: str | None = None, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to app instances reaching RUNNING status.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
app_key	str | None	Filter events for a specific app key.	None
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

async def on_app_running(
    self,
    *,
    handler: "HandlerType",
    app_key: str | None = None,
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to app instances reaching RUNNING status.

    Args:
        handler: The function to call when the event matches.
        app_key: Filter events for a specific app key.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener.
    """

    return await self.on_app_state_changed(
        handler=handler,
        app_key=app_key,
        status=ResourceStatus.RUNNING,
        where=where,
        kwargs=kwargs,
        name=name,
        **opts,
    )

on_app_stopping(*, handler: HandlerType, app_key: str | None = None, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription async

Subscribe to app instances entering STOPPING status.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
app_key	str | None	Filter events for a specific app key.	None
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/bus.py

async def on_app_stopping(
    self,
    *,
    handler: "HandlerType",
    app_key: str | None = None,
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to app instances entering STOPPING status.

    Args:
        handler: The function to call when the event matches.
        app_key: Filter events for a specific app key.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener.
    """

    return await self.on_app_state_changed(
        handler=handler,
        app_key=app_key,
        status=ResourceStatus.STOPPING,
        where=where,
        kwargs=kwargs,
        name=name,
        **opts,
    )

BusErrorContext dataclass

Bases: ErrorContext

Context passed to bus error handlers when a listener raises an exception.

Attributes:

Name	Type	Description
exception	BaseException	The exception that was raised by the listener. Retains its __traceback__ chain — use traceback (the string field) for display. The live traceback pins originating stack frame locals until this context is garbage-collected (bounded by error_handler_timeout_seconds).
traceback	str	Formatted traceback string. Always a non-empty string — the design explicitly requires always-populated tracebacks in the user-facing context, unlike the framework's own log suppression which may suppress them.
topic	str	The topic the listener was registered on.
listener_name	str	The name of the listener function that raised the exception.
event	Event[Any]	The event that was being processed when the exception occurred.

Source code in src/hassette/bus/error_context.py

@dataclass(frozen=True)
class BusErrorContext(ErrorContext):
    """Context passed to bus error handlers when a listener raises an exception.

    Attributes:
        exception: The exception that was raised by the listener. Retains its
            ``__traceback__`` chain — use ``traceback`` (the string field) for display.
            The live traceback pins originating stack frame locals until this context
            is garbage-collected (bounded by ``error_handler_timeout_seconds``).
        traceback: Formatted traceback string. Always a non-empty string — the
            design explicitly requires always-populated tracebacks in the user-facing
            context, unlike the framework's own log suppression which may suppress them.
        topic: The topic the listener was registered on.
        listener_name: The name of the listener function that raised the exception.
        event: The event that was being processed when the exception occurred.
    """

    topic: str
    listener_name: str
    event: "Event[Any]"

    @property
    def _domain_label(self) -> str:
        return f"topic={self.topic}, listener={self.listener_name}"

DurationConfig dataclass

Groups duration-hold configuration fields and owns the timer lifecycle; timer is attached via attach_timer().

Source code in src/hassette/bus/listeners.py

@dataclass(slots=True)
class DurationConfig:
    """Groups duration-hold configuration fields and owns the timer lifecycle; timer is attached via attach_timer()."""

    entity_id: str
    """Entity ID this duration listener is tracking. Required — non-empty."""

    duration: float | None = None
    """Duration in seconds the entity must remain in the matching state before the handler fires.
    None for immediate-only or entity_id-only listeners."""

    immediate: bool = False
    """If True, fire the handler immediately with the current entity state on registration."""

    is_attribute_listener: bool = False
    """True when this listener was registered via on_attribute_change."""

    hold_predicate: "Predicate | None" = None
    """State-value predicates only (excludes transition predicates like StateFrom, StateDidChange).
    Used by DurationTimer for cancel evaluation and fire-time recheck. None when not set."""

    _timer: "DurationTimer | None" = field(default=None, init=False)
    """Duration timer. Attached via attach_timer() during BusService registration."""

    def __post_init__(self) -> None:
        if not self.entity_id:
            raise ValueError("'entity_id' must be a non-empty string")
        if self.duration is not None and self.duration <= 0:
            raise ValueError("'duration' must be a positive number")

    @property
    def timer(self) -> "DurationTimer":
        """Return the attached DurationTimer. Asserts it has been attached."""
        assert self._timer is not None, "timer not yet attached — call attach_timer() first"
        return self._timer

    def cancel_timer(self) -> None:
        """Cancel the attached duration timer if present."""
        if self._timer is not None:
            self._timer.cancel()

    def attach_timer(
        self,
        task_bucket: "TaskBucket",
        owner_id: str,
        create_cancel_sub: "Callable[[], Subscription]",
        on_cancel: Callable[[], None] | None = None,
    ) -> None:
        """Construct a DurationTimer and store it.

        BusService calls this method during registration, passing the
        cancel-subscription factory and on_cancel callback. Counter
        ownership stays in BusService.
        """
        assert self._timer is None, "timer already attached — call cancel() before re-attaching"
        assert self.duration is not None, "attach_timer() requires a non-None duration"
        self._timer = DurationTimer(
            task_bucket=task_bucket,
            duration=self.duration,
            predicates=self.hold_predicate,
            entity_id=self.entity_id,
            owner_id=owner_id,
            create_cancel_sub=create_cancel_sub,
            on_cancel=on_cancel,
        )

entity_id: str instance-attribute

Entity ID this duration listener is tracking. Required — non-empty.

duration: float | None = None class-attribute instance-attribute

Duration in seconds the entity must remain in the matching state before the handler fires. None for immediate-only or entity_id-only listeners.

immediate: bool = False class-attribute instance-attribute

If True, fire the handler immediately with the current entity state on registration.

is_attribute_listener: bool = False class-attribute instance-attribute

True when this listener was registered via on_attribute_change.

hold_predicate: Predicate | None = None class-attribute instance-attribute

State-value predicates only (excludes transition predicates like StateFrom, StateDidChange). Used by DurationTimer for cancel evaluation and fire-time recheck. None when not set.

timer: DurationTimer property

Return the attached DurationTimer. Asserts it has been attached.

cancel_timer() -> None

Cancel the attached duration timer if present.

Source code in src/hassette/bus/listeners.py

def cancel_timer(self) -> None:
    """Cancel the attached duration timer if present."""
    if self._timer is not None:
        self._timer.cancel()

attach_timer(task_bucket: TaskBucket, owner_id: str, create_cancel_sub: Callable[[], Subscription], on_cancel: Callable[[], None] | None = None) -> None

Construct a DurationTimer and store it.

BusService calls this method during registration, passing the cancel-subscription factory and on_cancel callback. Counter ownership stays in BusService.

Source code in src/hassette/bus/listeners.py

def attach_timer(
    self,
    task_bucket: "TaskBucket",
    owner_id: str,
    create_cancel_sub: "Callable[[], Subscription]",
    on_cancel: Callable[[], None] | None = None,
) -> None:
    """Construct a DurationTimer and store it.

    BusService calls this method during registration, passing the
    cancel-subscription factory and on_cancel callback. Counter
    ownership stays in BusService.
    """
    assert self._timer is None, "timer already attached — call cancel() before re-attaching"
    assert self.duration is not None, "attach_timer() requires a non-None duration"
    self._timer = DurationTimer(
        task_bucket=task_bucket,
        duration=self.duration,
        predicates=self.hold_predicate,
        entity_id=self.entity_id,
        owner_id=owner_id,
        create_cancel_sub=create_cancel_sub,
        on_cancel=on_cancel,
    )

HandlerInvoker dataclass

Owns handler invocation, async wrapping, parameter injection, rate limiting, and the once-guard.

Source code in src/hassette/bus/listeners.py

@dataclass(slots=True)
class HandlerInvoker:
    """Owns handler invocation, async wrapping, parameter injection, rate limiting, and the once-guard."""

    orig_handler: "HandlerType"
    """Original handler function provided by the user."""

    async_handler: "AsyncHandlerType"
    """Async-wrapped handler function."""

    injector: ParameterInjector
    """Parameter injector for dependency injection."""

    kwargs: Mapping[str, Any] | None
    """Keyword arguments to pass to the handler."""

    error_handler: "BusErrorHandlerType | None"
    """Optional per-listener error handler."""

    app_error_handler_resolver: "Callable[[], BusErrorHandlerType | None] | None"
    """Closure that resolves the app-level error handler at dispatch time."""

    rate_limiter: RateLimiter | None
    """Rate limiter for debounce/throttle. None when no rate limiting is configured."""

    once: bool = False
    """Whether this invoker fires only once. Intentional copy of ListenerOptions.once —
    dispatch() needs this but cannot back-reference options without a circular dependency."""

    fired: bool = field(default=False, init=False)
    """Guard for once=True: set before the first invocation to prevent double-fire."""

    @classmethod
    def create(
        cls,
        task_bucket: "TaskBucket",
        handler: "HandlerType",
        kwargs: Mapping[str, Any] | None,
        options: ListenerOptions,
        error_handler: "BusErrorHandlerType | None" = None,
        app_error_handler_resolver: "Callable[[], BusErrorHandlerType | None] | None" = None,
    ) -> "HandlerInvoker":
        """Construct a HandlerInvoker from a handler and options.

        Builds the async wrapper, injector, and rate limiter. Copies options.once.

        Args:
            task_bucket: TaskBucket for async adapter and rate limiter.
            handler: The user-supplied handler callable.
            kwargs: Optional keyword arguments to pass to the handler.
            options: Behavioral options (once, debounce, throttle).
            error_handler: Optional per-listener error handler.
            app_error_handler_resolver: Closure for app-level error handler resolution.
        """
        handler_name = callable_name(handler)
        signature = get_typed_signature(handler)
        async_handler = make_async_handler(handler, task_bucket)
        injector = ParameterInjector(handler_name, signature)

        rate_limiter: RateLimiter | None = None
        if options.debounce is not None or options.throttle is not None:
            rate_limiter = RateLimiter(
                task_bucket=task_bucket,
                debounce=options.debounce,
                throttle=options.throttle,
                handler_name=handler_name,
            )

        return cls(
            orig_handler=handler,
            async_handler=async_handler,
            injector=injector,
            kwargs=kwargs,
            error_handler=error_handler,
            app_error_handler_resolver=app_error_handler_resolver,
            rate_limiter=rate_limiter,
            once=options.once,
        )

    def mark_fired(self) -> None:
        """Mark this once-invoker as having fired. Called by dispatch() and Listener.cancel()."""
        self.fired = True

    def set_app_error_handler_resolver(self, resolver: "Callable[[], BusErrorHandlerType | None]") -> None:
        """Set the closure that resolves the app-level error handler at dispatch time."""
        self.app_error_handler_resolver = resolver

    async def dispatch(self, invoke_fn: Callable[[], Awaitable[None]]) -> None:
        """Apply rate limiting around the given invoke function.

        BusService builds the invoke function (internal error-catching or tracked
        telemetry), HandlerInvoker wraps it with rate limiting. BusService never
        touches the RateLimiter directly.

        Includes once-guard: if ``once=True`` and the invoker has already fired,
        this method returns immediately. Safe without a lock — no ``await`` between
        check-and-set.
        """
        if self.once and self.fired:
            return
        if self.once:
            self.mark_fired()

        if self.rate_limiter:
            await self.rate_limiter.call(invoke_fn)
        else:
            await invoke_fn()

    def cancel(self) -> None:
        """Cancel any pending rate-limiter tasks."""
        if self.rate_limiter:
            self.rate_limiter.cancel()

    async def invoke(self, event: "Event[Any]") -> None:
        """Invoke the handler with dependency injection."""
        kwargs = self.injector.inject_parameters(event, **(self.kwargs or {}))
        await self.async_handler(**kwargs)

orig_handler: HandlerType instance-attribute

Original handler function provided by the user.

async_handler: AsyncHandlerType instance-attribute

Async-wrapped handler function.

injector: ParameterInjector instance-attribute

Parameter injector for dependency injection.

kwargs: Mapping[str, Any] | None instance-attribute

Keyword arguments to pass to the handler.

error_handler: BusErrorHandlerType | None instance-attribute

Optional per-listener error handler.

app_error_handler_resolver: Callable[[], BusErrorHandlerType | None] | None instance-attribute

Closure that resolves the app-level error handler at dispatch time.

rate_limiter: RateLimiter | None instance-attribute

Rate limiter for debounce/throttle. None when no rate limiting is configured.

once: bool = False class-attribute instance-attribute

Whether this invoker fires only once. Intentional copy of ListenerOptions.once — dispatch() needs this but cannot back-reference options without a circular dependency.

fired: bool = field(default=False, init=False) class-attribute instance-attribute

Guard for once=True: set before the first invocation to prevent double-fire.

create(task_bucket: TaskBucket, handler: HandlerType, kwargs: Mapping[str, Any] | None, options: ListenerOptions, error_handler: BusErrorHandlerType | None = None, app_error_handler_resolver: Callable[[], BusErrorHandlerType | None] | None = None) -> HandlerInvoker classmethod

Construct a HandlerInvoker from a handler and options.

Builds the async wrapper, injector, and rate limiter. Copies options.once.

Parameters:

Name	Type	Description	Default
task_bucket	TaskBucket	TaskBucket for async adapter and rate limiter.	required
handler	HandlerType	The user-supplied handler callable.	required
kwargs	Mapping[str, Any] | None	Optional keyword arguments to pass to the handler.	required
options	ListenerOptions	Behavioral options (once, debounce, throttle).	required
error_handler	BusErrorHandlerType | None	Optional per-listener error handler.	None
app_error_handler_resolver	Callable[[], BusErrorHandlerType | None] | None	Closure for app-level error handler resolution.	None

Source code in src/hassette/bus/listeners.py

@classmethod
def create(
    cls,
    task_bucket: "TaskBucket",
    handler: "HandlerType",
    kwargs: Mapping[str, Any] | None,
    options: ListenerOptions,
    error_handler: "BusErrorHandlerType | None" = None,
    app_error_handler_resolver: "Callable[[], BusErrorHandlerType | None] | None" = None,
) -> "HandlerInvoker":
    """Construct a HandlerInvoker from a handler and options.

    Builds the async wrapper, injector, and rate limiter. Copies options.once.

    Args:
        task_bucket: TaskBucket for async adapter and rate limiter.
        handler: The user-supplied handler callable.
        kwargs: Optional keyword arguments to pass to the handler.
        options: Behavioral options (once, debounce, throttle).
        error_handler: Optional per-listener error handler.
        app_error_handler_resolver: Closure for app-level error handler resolution.
    """
    handler_name = callable_name(handler)
    signature = get_typed_signature(handler)
    async_handler = make_async_handler(handler, task_bucket)
    injector = ParameterInjector(handler_name, signature)

    rate_limiter: RateLimiter | None = None
    if options.debounce is not None or options.throttle is not None:
        rate_limiter = RateLimiter(
            task_bucket=task_bucket,
            debounce=options.debounce,
            throttle=options.throttle,
            handler_name=handler_name,
        )

    return cls(
        orig_handler=handler,
        async_handler=async_handler,
        injector=injector,
        kwargs=kwargs,
        error_handler=error_handler,
        app_error_handler_resolver=app_error_handler_resolver,
        rate_limiter=rate_limiter,
        once=options.once,
    )

mark_fired() -> None

Mark this once-invoker as having fired. Called by dispatch() and Listener.cancel().

Source code in src/hassette/bus/listeners.py

def mark_fired(self) -> None:
    """Mark this once-invoker as having fired. Called by dispatch() and Listener.cancel()."""
    self.fired = True

set_app_error_handler_resolver(resolver: Callable[[], BusErrorHandlerType | None]) -> None

Set the closure that resolves the app-level error handler at dispatch time.

Source code in src/hassette/bus/listeners.py

def set_app_error_handler_resolver(self, resolver: "Callable[[], BusErrorHandlerType | None]") -> None:
    """Set the closure that resolves the app-level error handler at dispatch time."""
    self.app_error_handler_resolver = resolver

dispatch(invoke_fn: Callable[[], Awaitable[None]]) -> None async

Apply rate limiting around the given invoke function.

BusService builds the invoke function (internal error-catching or tracked telemetry), HandlerInvoker wraps it with rate limiting. BusService never touches the RateLimiter directly.

Includes once-guard: if once=True and the invoker has already fired, this method returns immediately. Safe without a lock — no await between check-and-set.

Source code in src/hassette/bus/listeners.py

async def dispatch(self, invoke_fn: Callable[[], Awaitable[None]]) -> None:
    """Apply rate limiting around the given invoke function.

    BusService builds the invoke function (internal error-catching or tracked
    telemetry), HandlerInvoker wraps it with rate limiting. BusService never
    touches the RateLimiter directly.

    Includes once-guard: if ``once=True`` and the invoker has already fired,
    this method returns immediately. Safe without a lock — no ``await`` between
    check-and-set.
    """
    if self.once and self.fired:
        return
    if self.once:
        self.mark_fired()

    if self.rate_limiter:
        await self.rate_limiter.call(invoke_fn)
    else:
        await invoke_fn()

cancel() -> None

Cancel any pending rate-limiter tasks.

Source code in src/hassette/bus/listeners.py

def cancel(self) -> None:
    """Cancel any pending rate-limiter tasks."""
    if self.rate_limiter:
        self.rate_limiter.cancel()

invoke(event: Event[Any]) -> None async

Invoke the handler with dependency injection.

Source code in src/hassette/bus/listeners.py

async def invoke(self, event: "Event[Any]") -> None:
    """Invoke the handler with dependency injection."""
    kwargs = self.injector.inject_parameters(event, **(self.kwargs or {}))
    await self.async_handler(**kwargs)

Listener dataclass

A listener for events with a specific topic and handler.

Composes four focused sub-structs (identity, invoker, options, duration_config) plus routing fields (topic, predicate) and minimal runtime state.

Source code in src/hassette/bus/listeners.py

@dataclass(slots=True)
class Listener:
    """A listener for events with a specific topic and handler.

    Composes four focused sub-structs (identity, invoker, options, duration_config)
    plus routing fields (topic, predicate) and minimal runtime state.
    """

    logger: Logger
    """Logger for the listener."""

    topic: str
    """Topic the listener is subscribed to."""

    predicate: "Predicate | None"
    """Predicate to filter events before invoking the handler."""

    identity: ListenerIdentity
    """Ownership and telemetry identity fields."""

    invoker: HandlerInvoker
    """Handler callable, dispatch engine, and once-guard."""

    options: ListenerOptions
    """Behavioral execution parameters."""

    duration_config: DurationConfig | None
    """Duration-hold configuration and timer. None for non-duration listeners."""

    listener_id: int = field(default_factory=next_id, init=False)
    """Unique identifier for the listener instance."""

    _cancelled: bool = field(default=False, init=False, repr=False)
    """Set by cancel() to signal that a pending add_listener task should skip route insertion."""

    db_id: int | None = field(default=None, init=False)
    """Database row ID for this listener. Set by the executor after persistence; None until then."""

    @property
    def is_cancelled(self) -> bool:
        """Whether this listener has been cancelled. Read-only — use cancel() to set."""
        return self._cancelled

    def mark_registered(self, db_id: int) -> None:
        """Set the database ID after persistence. One-time assignment by BusService."""
        if self.db_id is not None:
            LOGGER.warning(
                "Listener %s already registered with db_id=%s, ignoring new db_id=%s",
                self.listener_id,
                self.db_id,
                db_id,
            )
            return
        self.db_id = db_id

    def cancel(self) -> None:
        """Cancel the listener: set the cancelled flag and stop any pending tasks.

        Sets _cancelled flag, calls invoker.mark_fired() to prevent handler invocation
        on any in-flight dispatch task, and cancels rate limiter and duration timer.

        Terminal operation: the listener must not be reused after this call.
        """
        self._cancelled = True
        self.invoker.mark_fired()
        self.invoker.cancel()
        if self.duration_config is not None:
            self.duration_config.cancel_timer()

    def matches(self, ev: "Event[Any]") -> bool:
        """Check if the event matches the listener's predicate."""
        if self.predicate is None:
            return True
        matched = self.predicate(ev)

        verdict = "matched" if matched else "did not match"
        self.logger.debug("Listener %s %s predicate for event: %s", self, verdict, ev)
        return matched

    def __repr__(self) -> str:
        return f"Listener<{self.identity.owner_id} - {self.identity.handler_short_name}>"

    @classmethod
    def create(
        cls,
        topic: str,
        identity: ListenerIdentity,
        options: ListenerOptions,
        invoker: HandlerInvoker,
        where: "Predicate | Sequence[Predicate] | None" = None,
        duration_config: DurationConfig | None = None,
        logger: Logger = LOGGER,
    ) -> "Listener":
        """Create a Listener from pre-built sub-structs.

        Cross-concern validation (duration + debounce incompatibility) runs
        here since it spans two sub-structs.
        """
        if duration_config is not None and duration_config.duration is not None:
            if options.debounce is not None:
                raise ValueError("Cannot combine 'duration' with 'debounce'")
            if options.throttle is not None:
                raise ValueError("Cannot combine 'duration' with 'throttle'")

        pred = normalize_where(where)
        return cls(
            logger=logger,
            topic=topic,
            predicate=pred,
            identity=identity,
            invoker=invoker,
            options=options,
            duration_config=duration_config,
        )

    @classmethod
    def create_cancel_listener(
        cls,
        task_bucket: "TaskBucket",
        owner_id: str,
        topic: str,
        handler: "HandlerType",
        predicate: "Predicate | None" = None,
    ) -> "Listener":
        """Create a framework cancel-listener with sensible defaults.

        Produces a listener with source_tier='framework'. No rate limiter,
        no error handler, no duration config.
        """
        handler_name = callable_name(handler)
        short_name = callable_short_name(handler)

        identity = ListenerIdentity(
            owner_id=owner_id,
            handler_name=handler_name,
            handler_short_name=short_name,
            source_tier="framework",
        )

        options = ListenerOptions()

        invoker = HandlerInvoker.create(
            task_bucket=task_bucket,
            handler=handler,
            kwargs=None,
            options=options,
            error_handler=None,
        )

        return cls(
            logger=LOGGER,
            topic=topic,
            predicate=predicate,
            identity=identity,
            invoker=invoker,
            options=options,
            duration_config=None,
        )

logger: Logger instance-attribute

Logger for the listener.

topic: str instance-attribute

Topic the listener is subscribed to.

predicate: Predicate | None instance-attribute

Predicate to filter events before invoking the handler.

identity: ListenerIdentity instance-attribute

Ownership and telemetry identity fields.

invoker: HandlerInvoker instance-attribute

Handler callable, dispatch engine, and once-guard.

options: ListenerOptions instance-attribute

Behavioral execution parameters.

duration_config: DurationConfig | None instance-attribute

Duration-hold configuration and timer. None for non-duration listeners.

listener_id: int = field(default_factory=next_id, init=False) class-attribute instance-attribute

Unique identifier for the listener instance.

db_id: int | None = field(default=None, init=False) class-attribute instance-attribute

Database row ID for this listener. Set by the executor after persistence; None until then.

is_cancelled: bool property

Whether this listener has been cancelled. Read-only — use cancel() to set.

mark_registered(db_id: int) -> None

Set the database ID after persistence. One-time assignment by BusService.

Source code in src/hassette/bus/listeners.py

def mark_registered(self, db_id: int) -> None:
    """Set the database ID after persistence. One-time assignment by BusService."""
    if self.db_id is not None:
        LOGGER.warning(
            "Listener %s already registered with db_id=%s, ignoring new db_id=%s",
            self.listener_id,
            self.db_id,
            db_id,
        )
        return
    self.db_id = db_id

cancel() -> None

Cancel the listener: set the cancelled flag and stop any pending tasks.

Sets _cancelled flag, calls invoker.mark_fired() to prevent handler invocation on any in-flight dispatch task, and cancels rate limiter and duration timer.

Terminal operation: the listener must not be reused after this call.

Source code in src/hassette/bus/listeners.py

def cancel(self) -> None:
    """Cancel the listener: set the cancelled flag and stop any pending tasks.

    Sets _cancelled flag, calls invoker.mark_fired() to prevent handler invocation
    on any in-flight dispatch task, and cancels rate limiter and duration timer.

    Terminal operation: the listener must not be reused after this call.
    """
    self._cancelled = True
    self.invoker.mark_fired()
    self.invoker.cancel()
    if self.duration_config is not None:
        self.duration_config.cancel_timer()

matches(ev: Event[Any]) -> bool

Check if the event matches the listener's predicate.

Source code in src/hassette/bus/listeners.py

def matches(self, ev: "Event[Any]") -> bool:
    """Check if the event matches the listener's predicate."""
    if self.predicate is None:
        return True
    matched = self.predicate(ev)

    verdict = "matched" if matched else "did not match"
    self.logger.debug("Listener %s %s predicate for event: %s", self, verdict, ev)
    return matched

create(topic: str, identity: ListenerIdentity, options: ListenerOptions, invoker: HandlerInvoker, where: Predicate | Sequence[Predicate] | None = None, duration_config: DurationConfig | None = None, logger: Logger = LOGGER) -> Listener classmethod

Create a Listener from pre-built sub-structs.

Cross-concern validation (duration + debounce incompatibility) runs here since it spans two sub-structs.

Source code in src/hassette/bus/listeners.py

@classmethod
def create(
    cls,
    topic: str,
    identity: ListenerIdentity,
    options: ListenerOptions,
    invoker: HandlerInvoker,
    where: "Predicate | Sequence[Predicate] | None" = None,
    duration_config: DurationConfig | None = None,
    logger: Logger = LOGGER,
) -> "Listener":
    """Create a Listener from pre-built sub-structs.

    Cross-concern validation (duration + debounce incompatibility) runs
    here since it spans two sub-structs.
    """
    if duration_config is not None and duration_config.duration is not None:
        if options.debounce is not None:
            raise ValueError("Cannot combine 'duration' with 'debounce'")
        if options.throttle is not None:
            raise ValueError("Cannot combine 'duration' with 'throttle'")

    pred = normalize_where(where)
    return cls(
        logger=logger,
        topic=topic,
        predicate=pred,
        identity=identity,
        invoker=invoker,
        options=options,
        duration_config=duration_config,
    )

create_cancel_listener(task_bucket: TaskBucket, owner_id: str, topic: str, handler: HandlerType, predicate: Predicate | None = None) -> Listener classmethod

Create a framework cancel-listener with sensible defaults.

Produces a listener with source_tier='framework'. No rate limiter, no error handler, no duration config.

Source code in src/hassette/bus/listeners.py

@classmethod
def create_cancel_listener(
    cls,
    task_bucket: "TaskBucket",
    owner_id: str,
    topic: str,
    handler: "HandlerType",
    predicate: "Predicate | None" = None,
) -> "Listener":
    """Create a framework cancel-listener with sensible defaults.

    Produces a listener with source_tier='framework'. No rate limiter,
    no error handler, no duration config.
    """
    handler_name = callable_name(handler)
    short_name = callable_short_name(handler)

    identity = ListenerIdentity(
        owner_id=owner_id,
        handler_name=handler_name,
        handler_short_name=short_name,
        source_tier="framework",
    )

    options = ListenerOptions()

    invoker = HandlerInvoker.create(
        task_bucket=task_bucket,
        handler=handler,
        kwargs=None,
        options=options,
        error_handler=None,
    )

    return cls(
        logger=LOGGER,
        topic=topic,
        predicate=predicate,
        identity=identity,
        invoker=invoker,
        options=options,
        duration_config=None,
    )

ListenerIdentity dataclass

Groups ownership and telemetry fields that identify who registered a listener and where it came from.

Source code in src/hassette/bus/listeners.py

@dataclass(slots=True)
class ListenerIdentity:
    """Groups ownership and telemetry fields that identify who registered a listener and where it came from."""

    owner_id: str
    """Unique string identifier for the owner of the listener."""

    handler_name: str
    """Human-readable fully-qualified name for the handler, computed once at creation time."""

    handler_short_name: str
    """Short (last-segment) name for the handler, computed once at creation time."""

    app_key: str = ""
    """Configuration-level app key for DB registration (e.g., 'my_app'). Empty for non-App owners."""

    instance_index: int = 0
    """App instance index for DB registration. 0 for non-App owners."""

    name: str | None = None
    """Optional stable name for the listener (the name= escape hatch on Bus.on())."""

    source_tier: SourceTier = "app"
    """Whether this listener originates from a user app or the framework itself."""

    source_location: str = ""
    """Captured source location (file:line) of the user code that registered this listener."""

    registration_source: str = ""
    """Captured source code snippet of the registration call."""

owner_id: str instance-attribute

Unique string identifier for the owner of the listener.

handler_name: str instance-attribute

Human-readable fully-qualified name for the handler, computed once at creation time.

handler_short_name: str instance-attribute

Short (last-segment) name for the handler, computed once at creation time.

app_key: str = '' class-attribute instance-attribute

Configuration-level app key for DB registration (e.g., 'my_app'). Empty for non-App owners.

instance_index: int = 0 class-attribute instance-attribute

App instance index for DB registration. 0 for non-App owners.

name: str | None = None class-attribute instance-attribute

Optional stable name for the listener (the name= escape hatch on Bus.on()).

source_tier: SourceTier = 'app' class-attribute instance-attribute

Whether this listener originates from a user app or the framework itself.

source_location: str = '' class-attribute instance-attribute

Captured source location (file:line) of the user code that registered this listener.

registration_source: str = '' class-attribute instance-attribute

Captured source code snippet of the registration call.

ListenerOptions dataclass

Behavioral timing parameters (once, debounce, throttle, timeout, priority) with validation.

Source code in src/hassette/bus/listeners.py

@dataclass(slots=True)
class ListenerOptions:
    """Behavioral timing parameters (once, debounce, throttle, timeout, priority) with validation."""

    once: bool = False
    """Whether the listener should be removed after one invocation."""

    debounce: float | None = None
    """Debounce delay in seconds. Events reset the timer; handler fires after the quiet period."""

    throttle: float | None = None
    """Throttle interval in seconds. At most one handler execution per window; extras are dropped."""

    timeout: float | None = None
    """Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config.
    None means fall through to the config default."""

    timeout_disabled: bool = False
    """When True, disables timeout enforcement for this listener regardless of config."""

    priority: int = 0
    """Priority for listener ordering. Higher values run first. Default is 0 for app handlers."""

    def __post_init__(self) -> None:
        if self.debounce is not None and self.debounce <= 0:
            raise ValueError("'debounce' must be a positive number")
        if self.throttle is not None and self.throttle <= 0:
            raise ValueError("'throttle' must be a positive number")
        if self.debounce is not None and self.throttle is not None:
            raise ValueError("Cannot specify both 'debounce' and 'throttle' parameters")
        if self.once and (self.debounce is not None or self.throttle is not None):
            raise ValueError("Cannot combine 'once=True' with 'debounce' or 'throttle'")
        if self.timeout is not None and (isinstance(self.timeout, bool) or self.timeout <= 0):
            raise ValueError("timeout must be a positive number")
        if self.timeout_disabled and self.timeout is not None:
            raise ValueError("Cannot specify both 'timeout' and 'timeout_disabled=True'")

once: bool = False class-attribute instance-attribute

Whether the listener should be removed after one invocation.

debounce: float | None = None class-attribute instance-attribute

Debounce delay in seconds. Events reset the timer; handler fires after the quiet period.

throttle: float | None = None class-attribute instance-attribute

Throttle interval in seconds. At most one handler execution per window; extras are dropped.

timeout: float | None = None class-attribute instance-attribute

Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config. None means fall through to the config default.

timeout_disabled: bool = False class-attribute instance-attribute

When True, disables timeout enforcement for this listener regardless of config.

priority: int = 0 class-attribute instance-attribute

Priority for listener ordering. Higher values run first. Default is 0 for app handlers.

Subscription dataclass

A subscription to an event topic with a specific listener key.

This class is used to manage the lifecycle of a listener, allowing it to be cancelled or managed within a context.

Source code in src/hassette/bus/listeners.py

@dataclass(slots=True)
class Subscription:
    """A subscription to an event topic with a specific listener key.

    This class is used to manage the lifecycle of a listener, allowing it to be cancelled
    or managed within a context.
    """

    listener: Listener
    """The listener associated with this subscription."""

    unsubscribe: Callable[[], None]
    """Function to call to unsubscribe the listener."""

    def cancel(self) -> None:
        """Cancel the subscription by calling the unsubscribe function."""
        self.unsubscribe()

listener: Listener instance-attribute

The listener associated with this subscription.

unsubscribe: Callable[[], None] instance-attribute

Function to call to unsubscribe the listener.

cancel() -> None

Cancel the subscription by calling the unsubscribe function.

Source code in src/hassette/bus/listeners.py

def cancel(self) -> None:
    """Cancel the subscription by calling the unsubscribe function."""
    self.unsubscribe()

BusSyncFacade

Bases: Resource

Synchronous facade for the event bus.

This class provides synchronous methods that wrap the asynchronous registration methods of the Bus class, allowing listeners to be registered from synchronous code (for example, an AppSync lifecycle hook running in a worker thread).

These methods must not be called from within the event loop; doing so raises a RuntimeError. Use the asynchronous methods on Bus directly when operating within an event loop.

Source code in src/hassette/bus/sync.py

class BusSyncFacade(Resource):
    """Synchronous facade for the event bus.

    This class provides synchronous methods that wrap the asynchronous registration methods of
    the Bus class, allowing listeners to be registered from synchronous code (for example, an
    ``AppSync`` lifecycle hook running in a worker thread).

    These methods must not be called from within the event loop; doing so raises a RuntimeError.
    Use the asynchronous methods on ``Bus`` directly when operating within an event loop.
    """

    _bus: "Bus"

    def __init__(self, hassette: "Hassette", *, bus: "Bus", parent: Resource | None = None) -> None:
        super().__init__(hassette, parent=parent)
        self._bus = bus

    async def on_initialize(self) -> None:
        self.mark_ready(reason="Synchronous Bus facade initialized")

    @property
    def config_log_level(self) -> LOG_LEVEL_TYPE:
        return self.hassette.config.logging.bus_service

    def add_listener(self, listener: "Listener") -> None:
        """Add a pre-built listener to the bus.

        This is the direct entry point for callers that construct a ``Listener``
        externally. The normal registration flow (``on_state_change``, ``on()``,
        etc.) goes through ``_on_internal`` instead.

        Raises:
            ListenerNameRequiredError: If the listener has no ``name`` (required for all DB-registered
                listeners, including once-listeners; cancel-listeners bypass this path entirely).
            DuplicateListenerError: If the listener's natural key is already registered on this bus instance."""

        return self.task_bucket.run_sync(self._bus.add_listener(listener))

    def emit(self, topic: str, data: object) -> None:
        """Broadcast data to all subscribers of the given topic.

        Subscribers annotated with ``D.EventData[T]`` receive ``data`` pre-extracted.
        If the internal event stream is closed (during shutdown), the event is silently dropped."""

        return self.task_bucket.run_sync(self._bus.emit(topic, data))

    def on(
        self,
        *,
        topic: str,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        once: bool = False,
        debounce: float | None = None,
        throttle: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
    ) -> Subscription:
        """Subscribe to an event topic with optional filtering and modifiers.

        This is the public registration method for raw topic subscriptions.
        Registration completes before the call returns —
        ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            topic: The event topic to listen to.
            handler: The function to call when the event matches.
            where: Optional predicates to filter events. These can be custom callables or predefined predicates from
                `hassette.event_handling.predicates`. They will receive the full event for evaluation.
            kwargs: Keyword arguments to pass to the handler.
            once: If True, the handler will be called only once and then removed.
            debounce: If set, applies a debounce to the handler.
            throttle: If set, applies a throttle to the handler.
            timeout: Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config.
                None means fall through to the config default.
            timeout_disabled: When True, disables timeout enforcement for this listener regardless of config.
            name: Required. Stable string identifier for this listener. Forms part of the natural
                key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
                restarts. Omitting it raises ``ListenerNameRequiredError`` at call time.
            on_error: Optional per-listener error handler.

        Returns:
            A subscription object. ``sub.cancel()`` removes the listener.
            ``sub.listener.db_id`` is a valid integer immediately on return.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already registered."""

        return self.task_bucket.run_sync(
            self._bus.on(
                topic=topic,
                handler=handler,
                where=where,
                kwargs=kwargs,
                once=once,
                debounce=debounce,
                throttle=throttle,
                timeout=timeout,
                timeout_disabled=timeout_disabled,
                name=name,
                on_error=on_error,
            )
        )

    def on_state_change(
        self,
        entity_id: str,
        *,
        handler: "HandlerType",
        changed: bool | ComparisonCondition = True,
        changed_from: "ChangeType" = NOT_PROVIDED,
        changed_to: "ChangeType" = NOT_PROVIDED,
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        immediate: bool = False,
        duration: float | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to state changes for a specific entity.

        Registration — routing and database
        persistence — completes before the call returns. ``sub.listener.db_id`` is a valid
        integer immediately on return.

        Args:
            entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
            handler: The function to call when the event matches.
            changed: Whether to filter only events where the state changed. If a ComparisonCondition is provided, it
                will be used to compare the old and new state values.
            changed_from: A value or callable that will be used to filter state changes *from* this value.
            changed_to: A value or callable that will be used to filter state changes *to* this value.
            where: Additional predicates to filter events (e.g. ValueIs) or custom callables. These will receive the
                full event for evaluation.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. A stable string identifier for this listener. Forms part of the natural
                key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
                restarts. Omitting it raises ``ListenerNameRequiredError`` at call time.
            **opts: Additional options like `once`, `debounce` and `throttle`.

        Returns:
            A subscription object. ``sub.listener.db_id`` is set immediately. ``sub.cancel()``
            removes the listener from routing.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
                registered on this bus in the current session."""

        return self.task_bucket.run_sync(
            self._bus.on_state_change(
                entity_id,
                handler=handler,
                changed=changed,
                changed_from=changed_from,
                changed_to=changed_to,
                where=where,
                kwargs=kwargs,
                immediate=immediate,
                duration=duration,
                name=name,
                on_error=on_error,
                **opts,
            )
        )

    def on_attribute_change(
        self,
        entity_id: str,
        attr: str,
        *,
        handler: "HandlerType",
        changed: bool | ComparisonCondition = True,
        changed_from: "ChangeType" = NOT_PROVIDED,
        changed_to: "ChangeType" = NOT_PROVIDED,
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        immediate: bool = False,
        duration: float | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to state change events for a specific entity's attribute.

        Registration completes before the call
        returns. ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
            attr: The attribute name to filter changes on (e.g., "volume").
            handler: The function to call when the event matches.
            changed: Whether to filter only events where the attribute changed. If a ComparisonCondition is provided,
                it will be used to compare the old and new attribute values.
            changed_from: A value or callable that will be used to filter attribute changes *from* this value.
            changed_to: A value or callable that will be used to filter attribute changes *to* this value.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable string identifier. Omitting it raises ``ListenerNameRequiredError``.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object. ``sub.listener.db_id`` is set immediately.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already registered."""

        return self.task_bucket.run_sync(
            self._bus.on_attribute_change(
                entity_id,
                attr,
                handler=handler,
                changed=changed,
                changed_from=changed_from,
                changed_to=changed_to,
                where=where,
                kwargs=kwargs,
                immediate=immediate,
                duration=duration,
                name=name,
                on_error=on_error,
                **opts,
            )
        )

    def on_call_service(
        self,
        domain: str | None = None,
        service: str | None = None,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to service call events.

        Registration completes before the call
        returns. ``sub.listener.db_id`` is a valid integer immediately on return.

        Args:
            domain: The domain to filter service calls (e.g., "light").
            service: The service to filter service calls (e.g., "turn_on").
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. Stable string identifier for this listener. Omitting it raises
                ``ListenerNameRequiredError`` at call time.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object. ``sub.listener.db_id`` is set immediately.

        Raises:
            ListenerNameRequiredError: If ``name`` is not provided.
            DuplicateListenerError: If a listener with the same ``(name, topic)`` is already registered."""

        return self.task_bucket.run_sync(
            self._bus.on_call_service(
                domain, service, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_component_loaded(
        self,
        component: str | None = None,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to component loaded events.

        Args:
            component: The component to filter load events (e.g., "light").
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener."""

        return self.task_bucket.run_sync(
            self._bus.on_component_loaded(
                component, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_service_registered(
        self,
        domain: str | None = None,
        service: str | None = None,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to service registered events.

        Args:
            domain: The domain to filter service registrations (e.g., "light").
            service: The service to filter service registrations (e.g., "turn_on").
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener."""

        return self.task_bucket.run_sync(
            self._bus.on_service_registered(
                domain, service, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_homeassistant_restart(
        self,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to Home Assistant restart events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener."""

        return self.task_bucket.run_sync(self._bus.on_homeassistant_restart(handler, where, kwargs, name, **opts))

    def on_homeassistant_start(
        self,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to Home Assistant start events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener."""

        return self.task_bucket.run_sync(self._bus.on_homeassistant_start(handler, where, kwargs, name, **opts))

    def on_homeassistant_stop(
        self,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to Home Assistant stop events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener."""

        return self.task_bucket.run_sync(self._bus.on_homeassistant_stop(handler, where, kwargs, name, **opts))

    def on_hassette_service_status(
        self,
        status: ResourceStatus | None = None,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to hassette service status events.

        Args:
            status: The status to filter events (e.g., ResourceStatus.STARTED).
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. A stable string identifier for this listener. Forms part of the
                natural key ``(app_key, instance_index, name, topic)`` used for upsert
                deduplication across restarts. Omitting it raises ``ListenerNameRequiredError``
                at call time.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener."""

        return self.task_bucket.run_sync(
            self._bus.on_hassette_service_status(
                status, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
            )
        )

    def on_hassette_service_failed(
        self,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to hassette service failed events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Required. A stable string identifier for this listener. Forms part of the
                natural key ``(app_key, instance_index, name, topic)`` used for upsert
                deduplication across restarts. Omitting it raises ``ListenerNameRequiredError``
                at call time.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener."""

        return self.task_bucket.run_sync(
            self._bus.on_hassette_service_failed(handler=handler, where=where, kwargs=kwargs, name=name, **opts)
        )

    def on_hassette_service_crashed(
        self,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to hassette service crashed events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener."""

        return self.task_bucket.run_sync(
            self._bus.on_hassette_service_crashed(handler=handler, where=where, kwargs=kwargs, name=name, **opts)
        )

    def on_hassette_service_started(
        self,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to hassette service started events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener."""

        return self.task_bucket.run_sync(
            self._bus.on_hassette_service_started(handler=handler, where=where, kwargs=kwargs, name=name, **opts)
        )

    def on_websocket_connected(
        self,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to websocket connected events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener."""

        return self.task_bucket.run_sync(
            self._bus.on_websocket_connected(handler=handler, where=where, kwargs=kwargs, name=name, **opts)
        )

    def on_websocket_disconnected(
        self,
        *,
        handler: "HandlerType",
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to websocket disconnected events.

        Args:
            handler: The function to call when the event matches.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener."""

        return self.task_bucket.run_sync(
            self._bus.on_websocket_disconnected(handler=handler, where=where, kwargs=kwargs, name=name, **opts)
        )

    def on_app_state_changed(
        self,
        *,
        handler: "HandlerType",
        app_key: str | None = None,
        status: ResourceStatus | None = None,
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        on_error: "BusErrorHandlerType | None" = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to app instance state change events.

        Args:
            handler: The function to call when the event matches.
            app_key: Filter events for a specific app key.
            status: Filter events for a specific status.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener."""

        return self.task_bucket.run_sync(
            self._bus.on_app_state_changed(
                handler=handler,
                app_key=app_key,
                status=status,
                where=where,
                kwargs=kwargs,
                name=name,
                on_error=on_error,
                **opts,
            )
        )

    def on_app_running(
        self,
        *,
        handler: "HandlerType",
        app_key: str | None = None,
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to app instances reaching RUNNING status.

        Args:
            handler: The function to call when the event matches.
            app_key: Filter events for a specific app key.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener."""

        return self.task_bucket.run_sync(
            self._bus.on_app_running(handler=handler, app_key=app_key, where=where, kwargs=kwargs, name=name, **opts)
        )

    def on_app_stopping(
        self,
        *,
        handler: "HandlerType",
        app_key: str | None = None,
        where: "Predicate | Sequence[Predicate] | None" = None,
        kwargs: Mapping[str, Any] | None = None,
        name: str | None = None,
        **opts: Unpack[Options],
    ) -> Subscription:
        """Subscribe to app instances entering STOPPING status.

        Args:
            handler: The function to call when the event matches.
            app_key: Filter events for a specific app key.
            where: Additional predicates to filter events.
            kwargs: Keyword arguments to pass to the handler.
            name: Stable name for this listener. Required on all DB-registered listeners.
            **opts: Additional options like `once`, `debounce`, and `throttle`.

        Returns:
            A subscription object that can be used to manage the listener."""

        return self.task_bucket.run_sync(
            self._bus.on_app_stopping(handler=handler, app_key=app_key, where=where, kwargs=kwargs, name=name, **opts)
        )

    def on_error(self, handler: "BusErrorHandlerType") -> None:
        """Register an app-level error handler for this bus.

        The handler is called when any listener on this bus raises an exception
        (including ``TimeoutError``) and the listener does not have its own
        per-registration error handler.

        This is an app-level fallback — it is resolved at dispatch time, not at listener
        registration time. A later call to ``on_error()`` replaces any previously registered
        handler.

        Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near
        app shutdown may be cancelled before they complete. Do not rely on error handlers
        for delivery-critical alerting during system teardown.

        Args:
            handler: A sync or async callable that accepts a :class:`~hassette.bus.error_context.BusErrorContext`."""

        return self._bus.on_error(handler)

    def remove_listener(self, listener: "Listener") -> None:
        """Remove a listener from the bus."""

        return self._bus.remove_listener(listener)

    def remove_all_listeners(self) -> None:
        """Remove all listeners owned by this bus's owner."""

        return self._bus.remove_all_listeners()

    def get_listeners(self) -> list["Listener"]:
        """Get all listeners owned by this bus's owner."""

        return self._bus.get_listeners()

add_listener(listener: Listener) -> None

Add a pre-built listener to the bus.

This is the direct entry point for callers that construct a Listener externally. The normal registration flow (on_state_change, on(), etc.) goes through _on_internal instead.

Raises:

Type	Description
ListenerNameRequiredError	If the listener has no name (required for all DB-registered listeners, including once-listeners; cancel-listeners bypass this path entirely).
DuplicateListenerError	If the listener's natural key is already registered on this bus instance.

Source code in src/hassette/bus/sync.py

def add_listener(self, listener: "Listener") -> None:
    """Add a pre-built listener to the bus.

    This is the direct entry point for callers that construct a ``Listener``
    externally. The normal registration flow (``on_state_change``, ``on()``,
    etc.) goes through ``_on_internal`` instead.

    Raises:
        ListenerNameRequiredError: If the listener has no ``name`` (required for all DB-registered
            listeners, including once-listeners; cancel-listeners bypass this path entirely).
        DuplicateListenerError: If the listener's natural key is already registered on this bus instance."""

    return self.task_bucket.run_sync(self._bus.add_listener(listener))

emit(topic: str, data: object) -> None

Broadcast data to all subscribers of the given topic.

Subscribers annotated with D.EventData[T] receive data pre-extracted. If the internal event stream is closed (during shutdown), the event is silently dropped.

Source code in src/hassette/bus/sync.py

def emit(self, topic: str, data: object) -> None:
    """Broadcast data to all subscribers of the given topic.

    Subscribers annotated with ``D.EventData[T]`` receive ``data`` pre-extracted.
    If the internal event stream is closed (during shutdown), the event is silently dropped."""

    return self.task_bucket.run_sync(self._bus.emit(topic, data))

on(*, topic: str, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, once: bool = False, debounce: float | None = None, throttle: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, name: str | None = None, on_error: BusErrorHandlerType | None = None) -> Subscription

Subscribe to an event topic with optional filtering and modifiers.

This is the public registration method for raw topic subscriptions. Registration completes before the call returns — sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
topic	str	The event topic to listen to.	required
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Optional predicates to filter events. These can be custom callables or predefined predicates from hassette.event_handling.predicates. They will receive the full event for evaluation.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
once	bool	If True, the handler will be called only once and then removed.	False
debounce	float | None	If set, applies a debounce to the handler.	None
throttle	float | None	If set, applies a throttle to the handler.	None
timeout	float | None	Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config. None means fall through to the config default.	None
timeout_disabled	bool	When True, disables timeout enforcement for this listener regardless of config.	False
name	str | None	Required. Stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it raises ListenerNameRequiredError at call time.	None
on_error	BusErrorHandlerType | None	Optional per-listener error handler.	None

Returns:

Type	Description
Subscription	A subscription object. sub.cancel() removes the listener.
Subscription	sub.listener.db_id is a valid integer immediately on return.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered.

Source code in src/hassette/bus/sync.py

def on(
    self,
    *,
    topic: str,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    once: bool = False,
    debounce: float | None = None,
    throttle: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
) -> Subscription:
    """Subscribe to an event topic with optional filtering and modifiers.

    This is the public registration method for raw topic subscriptions.
    Registration completes before the call returns —
    ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        topic: The event topic to listen to.
        handler: The function to call when the event matches.
        where: Optional predicates to filter events. These can be custom callables or predefined predicates from
            `hassette.event_handling.predicates`. They will receive the full event for evaluation.
        kwargs: Keyword arguments to pass to the handler.
        once: If True, the handler will be called only once and then removed.
        debounce: If set, applies a debounce to the handler.
        throttle: If set, applies a throttle to the handler.
        timeout: Per-listener timeout in seconds. Overrides the global event_handler_timeout_seconds config.
            None means fall through to the config default.
        timeout_disabled: When True, disables timeout enforcement for this listener regardless of config.
        name: Required. Stable string identifier for this listener. Forms part of the natural
            key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
            restarts. Omitting it raises ``ListenerNameRequiredError`` at call time.
        on_error: Optional per-listener error handler.

    Returns:
        A subscription object. ``sub.cancel()`` removes the listener.
        ``sub.listener.db_id`` is a valid integer immediately on return.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already registered."""

    return self.task_bucket.run_sync(
        self._bus.on(
            topic=topic,
            handler=handler,
            where=where,
            kwargs=kwargs,
            once=once,
            debounce=debounce,
            throttle=throttle,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            name=name,
            on_error=on_error,
        )
    )

on_state_change(entity_id: str, *, handler: HandlerType, changed: bool | ComparisonCondition = True, changed_from: ChangeType = NOT_PROVIDED, changed_to: ChangeType = NOT_PROVIDED, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, immediate: bool = False, duration: float | None = None, name: str | None = None, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to state changes for a specific entity.

Registration — routing and database persistence — completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
entity_id	str	The entity ID to filter events for (e.g., "media_player.living_room_speaker").	required
handler	HandlerType	The function to call when the event matches.	required
changed	bool | ComparisonCondition	Whether to filter only events where the state changed. If a ComparisonCondition is provided, it will be used to compare the old and new state values.	True
changed_from	ChangeType	A value or callable that will be used to filter state changes from this value.	NOT_PROVIDED
changed_to	ChangeType	A value or callable that will be used to filter state changes to this value.	NOT_PROVIDED
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events (e.g. ValueIs) or custom callables. These will receive the full event for evaluation.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Required. A stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it raises ListenerNameRequiredError at call time.	None
**opts	Unpack[Options]	Additional options like once, debounce and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object. sub.listener.db_id is set immediately. sub.cancel()
Subscription	removes the listener from routing.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered on this bus in the current session.

Source code in src/hassette/bus/sync.py

def on_state_change(
    self,
    entity_id: str,
    *,
    handler: "HandlerType",
    changed: bool | ComparisonCondition = True,
    changed_from: "ChangeType" = NOT_PROVIDED,
    changed_to: "ChangeType" = NOT_PROVIDED,
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    immediate: bool = False,
    duration: float | None = None,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to state changes for a specific entity.

    Registration — routing and database
    persistence — completes before the call returns. ``sub.listener.db_id`` is a valid
    integer immediately on return.

    Args:
        entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
        handler: The function to call when the event matches.
        changed: Whether to filter only events where the state changed. If a ComparisonCondition is provided, it
            will be used to compare the old and new state values.
        changed_from: A value or callable that will be used to filter state changes *from* this value.
        changed_to: A value or callable that will be used to filter state changes *to* this value.
        where: Additional predicates to filter events (e.g. ValueIs) or custom callables. These will receive the
            full event for evaluation.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. A stable string identifier for this listener. Forms part of the natural
            key ``(app_key, instance_index, name, topic)`` used for upsert deduplication across
            restarts. Omitting it raises ``ListenerNameRequiredError`` at call time.
        **opts: Additional options like `once`, `debounce` and `throttle`.

    Returns:
        A subscription object. ``sub.listener.db_id`` is set immediately. ``sub.cancel()``
        removes the listener from routing.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already
            registered on this bus in the current session."""

    return self.task_bucket.run_sync(
        self._bus.on_state_change(
            entity_id,
            handler=handler,
            changed=changed,
            changed_from=changed_from,
            changed_to=changed_to,
            where=where,
            kwargs=kwargs,
            immediate=immediate,
            duration=duration,
            name=name,
            on_error=on_error,
            **opts,
        )
    )

on_attribute_change(entity_id: str, attr: str, *, handler: HandlerType, changed: bool | ComparisonCondition = True, changed_from: ChangeType = NOT_PROVIDED, changed_to: ChangeType = NOT_PROVIDED, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, immediate: bool = False, duration: float | None = None, name: str | None = None, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to state change events for a specific entity's attribute.

Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
entity_id	str	The entity ID to filter events for (e.g., "media_player.living_room_speaker").	required
attr	str	The attribute name to filter changes on (e.g., "volume").	required
handler	HandlerType	The function to call when the event matches.	required
changed	bool | ComparisonCondition	Whether to filter only events where the attribute changed. If a ComparisonCondition is provided, it will be used to compare the old and new attribute values.	True
changed_from	ChangeType	A value or callable that will be used to filter attribute changes from this value.	NOT_PROVIDED
changed_to	ChangeType	A value or callable that will be used to filter attribute changes to this value.	NOT_PROVIDED
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Required. Stable string identifier. Omitting it raises ListenerNameRequiredError.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object. sub.listener.db_id is set immediately.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered.

Source code in src/hassette/bus/sync.py

def on_attribute_change(
    self,
    entity_id: str,
    attr: str,
    *,
    handler: "HandlerType",
    changed: bool | ComparisonCondition = True,
    changed_from: "ChangeType" = NOT_PROVIDED,
    changed_to: "ChangeType" = NOT_PROVIDED,
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    immediate: bool = False,
    duration: float | None = None,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to state change events for a specific entity's attribute.

    Registration completes before the call
    returns. ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        entity_id: The entity ID to filter events for (e.g., "media_player.living_room_speaker").
        attr: The attribute name to filter changes on (e.g., "volume").
        handler: The function to call when the event matches.
        changed: Whether to filter only events where the attribute changed. If a ComparisonCondition is provided,
            it will be used to compare the old and new attribute values.
        changed_from: A value or callable that will be used to filter attribute changes *from* this value.
        changed_to: A value or callable that will be used to filter attribute changes *to* this value.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable string identifier. Omitting it raises ``ListenerNameRequiredError``.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object. ``sub.listener.db_id`` is set immediately.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already registered."""

    return self.task_bucket.run_sync(
        self._bus.on_attribute_change(
            entity_id,
            attr,
            handler=handler,
            changed=changed,
            changed_from=changed_from,
            changed_to=changed_to,
            where=where,
            kwargs=kwargs,
            immediate=immediate,
            duration=duration,
            name=name,
            on_error=on_error,
            **opts,
        )
    )

on_call_service(domain: str | None = None, service: str | None = None, *, handler: HandlerType, where: Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to service call events.

Registration completes before the call returns. sub.listener.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
domain	str | None	The domain to filter service calls (e.g., "light").	None
service	str | None	The service to filter service calls (e.g., "turn_on").	None
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Required. Stable string identifier for this listener. Omitting it raises ListenerNameRequiredError at call time.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object. sub.listener.db_id is set immediately.

Raises:

Type	Description
ListenerNameRequiredError	If name is not provided.
DuplicateListenerError	If a listener with the same (name, topic) is already registered.

Source code in src/hassette/bus/sync.py

def on_call_service(
    self,
    domain: str | None = None,
    service: str | None = None,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | Mapping[str, ChangeType] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to service call events.

    Registration completes before the call
    returns. ``sub.listener.db_id`` is a valid integer immediately on return.

    Args:
        domain: The domain to filter service calls (e.g., "light").
        service: The service to filter service calls (e.g., "turn_on").
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. Stable string identifier for this listener. Omitting it raises
            ``ListenerNameRequiredError`` at call time.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object. ``sub.listener.db_id`` is set immediately.

    Raises:
        ListenerNameRequiredError: If ``name`` is not provided.
        DuplicateListenerError: If a listener with the same ``(name, topic)`` is already registered."""

    return self.task_bucket.run_sync(
        self._bus.on_call_service(
            domain, service, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_component_loaded(component: str | None = None, *, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to component loaded events.

Parameters:

Name	Type	Description	Default
component	str | None	The component to filter load events (e.g., "light").	None
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_component_loaded(
    self,
    component: str | None = None,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to component loaded events.

    Args:
        component: The component to filter load events (e.g., "light").
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener."""

    return self.task_bucket.run_sync(
        self._bus.on_component_loaded(
            component, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_service_registered(domain: str | None = None, service: str | None = None, *, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to service registered events.

Parameters:

Name	Type	Description	Default
domain	str | None	The domain to filter service registrations (e.g., "light").	None
service	str | None	The service to filter service registrations (e.g., "turn_on").	None
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_service_registered(
    self,
    domain: str | None = None,
    service: str | None = None,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to service registered events.

    Args:
        domain: The domain to filter service registrations (e.g., "light").
        service: The service to filter service registrations (e.g., "turn_on").
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener."""

    return self.task_bucket.run_sync(
        self._bus.on_service_registered(
            domain, service, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_homeassistant_restart(handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to Home Assistant restart events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_homeassistant_restart(
    self,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to Home Assistant restart events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener."""

    return self.task_bucket.run_sync(self._bus.on_homeassistant_restart(handler, where, kwargs, name, **opts))

on_homeassistant_start(handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to Home Assistant start events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_homeassistant_start(
    self,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to Home Assistant start events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener."""

    return self.task_bucket.run_sync(self._bus.on_homeassistant_start(handler, where, kwargs, name, **opts))

on_homeassistant_stop(handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to Home Assistant stop events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_homeassistant_stop(
    self,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to Home Assistant stop events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener."""

    return self.task_bucket.run_sync(self._bus.on_homeassistant_stop(handler, where, kwargs, name, **opts))

on_hassette_service_status(status: ResourceStatus | None = None, *, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to hassette service status events.

Parameters:

Name	Type	Description	Default
status	ResourceStatus | None	The status to filter events (e.g., ResourceStatus.STARTED).	None
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Required. A stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it raises ListenerNameRequiredError at call time.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_hassette_service_status(
    self,
    status: ResourceStatus | None = None,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to hassette service status events.

    Args:
        status: The status to filter events (e.g., ResourceStatus.STARTED).
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. A stable string identifier for this listener. Forms part of the
            natural key ``(app_key, instance_index, name, topic)`` used for upsert
            deduplication across restarts. Omitting it raises ``ListenerNameRequiredError``
            at call time.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener."""

    return self.task_bucket.run_sync(
        self._bus.on_hassette_service_status(
            status, handler=handler, where=where, kwargs=kwargs, name=name, on_error=on_error, **opts
        )
    )

on_hassette_service_failed(*, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to hassette service failed events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Required. A stable string identifier for this listener. Forms part of the natural key (app_key, instance_index, name, topic) used for upsert deduplication across restarts. Omitting it raises ListenerNameRequiredError at call time.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_hassette_service_failed(
    self,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to hassette service failed events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Required. A stable string identifier for this listener. Forms part of the
            natural key ``(app_key, instance_index, name, topic)`` used for upsert
            deduplication across restarts. Omitting it raises ``ListenerNameRequiredError``
            at call time.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener."""

    return self.task_bucket.run_sync(
        self._bus.on_hassette_service_failed(handler=handler, where=where, kwargs=kwargs, name=name, **opts)
    )

on_hassette_service_crashed(*, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to hassette service crashed events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_hassette_service_crashed(
    self,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to hassette service crashed events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener."""

    return self.task_bucket.run_sync(
        self._bus.on_hassette_service_crashed(handler=handler, where=where, kwargs=kwargs, name=name, **opts)
    )

on_hassette_service_started(*, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to hassette service started events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_hassette_service_started(
    self,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to hassette service started events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener."""

    return self.task_bucket.run_sync(
        self._bus.on_hassette_service_started(handler=handler, where=where, kwargs=kwargs, name=name, **opts)
    )

on_websocket_connected(*, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to websocket connected events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_websocket_connected(
    self,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to websocket connected events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener."""

    return self.task_bucket.run_sync(
        self._bus.on_websocket_connected(handler=handler, where=where, kwargs=kwargs, name=name, **opts)
    )

on_websocket_disconnected(*, handler: HandlerType, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to websocket disconnected events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_websocket_disconnected(
    self,
    *,
    handler: "HandlerType",
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to websocket disconnected events.

    Args:
        handler: The function to call when the event matches.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener."""

    return self.task_bucket.run_sync(
        self._bus.on_websocket_disconnected(handler=handler, where=where, kwargs=kwargs, name=name, **opts)
    )

on_app_state_changed(*, handler: HandlerType, app_key: str | None = None, status: ResourceStatus | None = None, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, on_error: BusErrorHandlerType | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to app instance state change events.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
app_key	str | None	Filter events for a specific app key.	None
status	ResourceStatus | None	Filter events for a specific status.	None
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_app_state_changed(
    self,
    *,
    handler: "HandlerType",
    app_key: str | None = None,
    status: ResourceStatus | None = None,
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    on_error: "BusErrorHandlerType | None" = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to app instance state change events.

    Args:
        handler: The function to call when the event matches.
        app_key: Filter events for a specific app key.
        status: Filter events for a specific status.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener."""

    return self.task_bucket.run_sync(
        self._bus.on_app_state_changed(
            handler=handler,
            app_key=app_key,
            status=status,
            where=where,
            kwargs=kwargs,
            name=name,
            on_error=on_error,
            **opts,
        )
    )

on_app_running(*, handler: HandlerType, app_key: str | None = None, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to app instances reaching RUNNING status.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
app_key	str | None	Filter events for a specific app key.	None
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_app_running(
    self,
    *,
    handler: "HandlerType",
    app_key: str | None = None,
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to app instances reaching RUNNING status.

    Args:
        handler: The function to call when the event matches.
        app_key: Filter events for a specific app key.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener."""

    return self.task_bucket.run_sync(
        self._bus.on_app_running(handler=handler, app_key=app_key, where=where, kwargs=kwargs, name=name, **opts)
    )

on_app_stopping(*, handler: HandlerType, app_key: str | None = None, where: Predicate | Sequence[Predicate] | None = None, kwargs: Mapping[str, Any] | None = None, name: str | None = None, **opts: Unpack[Options]) -> Subscription

Subscribe to app instances entering STOPPING status.

Parameters:

Name	Type	Description	Default
handler	HandlerType	The function to call when the event matches.	required
app_key	str | None	Filter events for a specific app key.	None
where	Predicate | Sequence[Predicate] | None	Additional predicates to filter events.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the handler.	None
name	str | None	Stable name for this listener. Required on all DB-registered listeners.	None
**opts	Unpack[Options]	Additional options like once, debounce, and throttle.	{}

Returns:

Type	Description
Subscription	A subscription object that can be used to manage the listener.

Source code in src/hassette/bus/sync.py

def on_app_stopping(
    self,
    *,
    handler: "HandlerType",
    app_key: str | None = None,
    where: "Predicate | Sequence[Predicate] | None" = None,
    kwargs: Mapping[str, Any] | None = None,
    name: str | None = None,
    **opts: Unpack[Options],
) -> Subscription:
    """Subscribe to app instances entering STOPPING status.

    Args:
        handler: The function to call when the event matches.
        app_key: Filter events for a specific app key.
        where: Additional predicates to filter events.
        kwargs: Keyword arguments to pass to the handler.
        name: Stable name for this listener. Required on all DB-registered listeners.
        **opts: Additional options like `once`, `debounce`, and `throttle`.

    Returns:
        A subscription object that can be used to manage the listener."""

    return self.task_bucket.run_sync(
        self._bus.on_app_stopping(handler=handler, app_key=app_key, where=where, kwargs=kwargs, name=name, **opts)
    )

on_error(handler: BusErrorHandlerType) -> None

Register an app-level error handler for this bus.

The handler is called when any listener on this bus raises an exception (including TimeoutError) and the listener does not have its own per-registration error handler.

This is an app-level fallback — it is resolved at dispatch time, not at listener registration time. A later call to on_error() replaces any previously registered handler.

Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near app shutdown may be cancelled before they complete. Do not rely on error handlers for delivery-critical alerting during system teardown.

Parameters:

Name	Type	Description	Default
handler	BusErrorHandlerType	A sync or async callable that accepts a :class:~hassette.bus.error_context.BusErrorContext.	required

Source code in src/hassette/bus/sync.py

def on_error(self, handler: "BusErrorHandlerType") -> None:
    """Register an app-level error handler for this bus.

    The handler is called when any listener on this bus raises an exception
    (including ``TimeoutError``) and the listener does not have its own
    per-registration error handler.

    This is an app-level fallback — it is resolved at dispatch time, not at listener
    registration time. A later call to ``on_error()`` replaces any previously registered
    handler.

    Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near
    app shutdown may be cancelled before they complete. Do not rely on error handlers
    for delivery-critical alerting during system teardown.

    Args:
        handler: A sync or async callable that accepts a :class:`~hassette.bus.error_context.BusErrorContext`."""

    return self._bus.on_error(handler)

remove_listener(listener: Listener) -> None

Remove a listener from the bus.

Source code in src/hassette/bus/sync.py

def remove_listener(self, listener: "Listener") -> None:
    """Remove a listener from the bus."""

    return self._bus.remove_listener(listener)

remove_all_listeners() -> None

Remove all listeners owned by this bus's owner.

Source code in src/hassette/bus/sync.py

def remove_all_listeners(self) -> None:
    """Remove all listeners owned by this bus's owner."""

    return self._bus.remove_all_listeners()

get_listeners() -> list[Listener]

Get all listeners owned by this bus's owner.

Source code in src/hassette/bus/sync.py

def get_listeners(self) -> list["Listener"]:
    """Get all listeners owned by this bus's owner."""

    return self._bus.get_listeners()