Listeners

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."""

    instance_name: str | None = None
    """App instance name, precomputed at registration time from the owning app's app_config.instance_name.
    None for framework-tier listeners and 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.

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

App instance name, precomputed at registration time from the owning app's app_config.instance_name. None for framework-tier listeners and 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."""

    mode: ExecutionMode = ExecutionMode.SINGLE
    """Overlap behavior when a trigger fires while a prior invocation still runs.

    ``single`` drops the re-fire, ``restart`` cancels-and-replaces, ``queued`` serializes,
    ``parallel`` runs concurrently. The tier-aware default (framework→parallel, app→single)
    is applied by the registration path, not by this dataclass default. Suppressed/dropped
    counts are live-only diagnostics held on the per-listener guard, reset on restart.
    """

    backpressure: BackpressurePolicy = BackpressurePolicy.BLOCK
    """Saturation policy for this listener when the dispatch concurrency semaphore is at capacity.

    ``block`` (default) waits for a slot, preserving today's behavior unchanged.
    ``drop_newest`` skips the event immediately when the bus is saturated — the handler is not
    invoked and one drop is recorded. Acts at the dispatch acquire gate, orthogonal to
    ``mode``/``debounce``/``throttle`` which act inside the invoker.
    """

    def __post_init__(self) -> None:
        # Coerce a raw string mode (arriving via the Options TypedDict or str ergonomics) into
        # the enum. An unknown value fails coercion — surface it as a clear ValueError.
        if not isinstance(self.mode, ExecutionMode):
            try:
                self.mode = ExecutionMode(self.mode)
            except ValueError as exc:
                valid = ", ".join(repr(m.value) for m in ExecutionMode)
                raise ValueError(f"Invalid execution mode {self.mode!r}; must be one of {valid}") from exc
        # Coerce a raw string backpressure policy the same way.
        if not isinstance(self.backpressure, BackpressurePolicy):
            try:
                self.backpressure = BackpressurePolicy(self.backpressure)
            except ValueError as exc:
                valid = ", ".join(repr(m.value) for m in BackpressurePolicy)
                raise ValueError(f"Invalid backpressure policy {self.backpressure!r}; must be one of {valid}") from exc
        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.

mode: ExecutionMode = ExecutionMode.SINGLE class-attribute instance-attribute

Overlap behavior when a trigger fires while a prior invocation still runs.

single drops the re-fire, restart cancels-and-replaces, queued serializes, parallel runs concurrently. The tier-aware default (framework→parallel, app→single) is applied by the registration path, not by this dataclass default. Suppressed/dropped counts are live-only diagnostics held on the per-listener guard, reset on restart.

backpressure: BackpressurePolicy = BackpressurePolicy.BLOCK class-attribute instance-attribute

Saturation policy for this listener when the dispatch concurrency semaphore is at capacity.

block (default) waits for a slot, preserving today's behavior unchanged. drop_newest skips the event immediately when the bus is saturated — the handler is not invoked and one drop is recorded. Acts at the dispatch acquire gate, orthogonal to mode/debounce/throttle which act inside the invoker.

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."""

    task_bucket: "TaskBucket"
    """TaskBucket used to spawn the cancellable child handler task for non-parallel modes."""

    guard: ExecutionModeGuard
    """Per-listener overlap state machine. Owns the running-task reference and live counters."""

    mode: ExecutionMode
    """Resolved overlap mode. Copied from options so dispatch can branch without reading the guard's
    private state."""

    handler_short_name: str
    """Short handler name, captured for the stall WARNING and debug logs."""

    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."""

    backpressure_dropped: int = 0
    """Count of events dropped at the dispatch acquire gate due to DROP_NEWEST backpressure.
    Incremented only in BusService.dispatch — one writer, on the event loop, no await between
    the locked() check and the increment. Live-only; resets on restart, never persisted."""

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

    pending_done: "set[asyncio.Future[None]]" = field(default_factory=set, init=False)
    """Unresolved per-invocation completion futures for non-parallel modes.

    Each ``run_with_mode`` call (single/restart/queued) parks its outer dispatch task on a future
    that resolves when the handler actually runs (or is dropped/released). A queued trigger accepted
    into the deque has no live child until drain time, so its future would hang forever if the
    listener is released first. ``release_guard`` resolves every remaining future here so those outer
    dispatch tasks unwind and ``_dispatch_pending`` settles."""

    @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,
            task_bucket=task_bucket,
            guard=ExecutionModeGuard(options.mode),
            mode=options.mode,
            handler_short_name=callable_short_name(handler),
            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 the once-guard, rate limiter, and overlap mode around the given invoke function.

        Order: once-guard → rate limiter (whether to start) → mode guard (overlap of started
        invocations). BusService builds ``invoke_fn`` (tracked telemetry); HandlerInvoker wraps it.

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

        Mode guard: ``parallel`` is a pass-through that awaits ``invoke_fn`` inline (byte-for-byte
        today's behavior, no child task). For ``single``/``restart``/``queued`` the guard receives a
        run-and-track callable that spawns a fresh child task through ``task_bucket``; this method
        then awaits that child so the outer dispatch task stays pending and remains counted by
        ``_dispatch_pending``. A ``restart`` cancellation of the child surfaces as ``CancelledError``
        inside the child only — it is swallowed here so the outer dispatch task does not crash.
        """
        if self.once and self.fired:
            return
        if self.once:
            self.mark_fired()

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

    async def run_with_mode(self, invoke_fn: Callable[[], Awaitable[None]]) -> None:
        """Apply the overlap mode guard to a single started invocation."""
        if self.mode is ExecutionMode.PARALLEL:
            await invoke_fn()
            return

        await run_through_guard(
            guard=self.guard,
            spawn=lambda coro, *, name: self.task_bucket.spawn(coro, name=name),
            pending_done=self.pending_done,
            invoke=invoke_fn,
            warn=self.warn_stalled,
            spawn_name="bus:mode_invocation",
            threshold=STALL_THRESHOLD_SECONDS,
        )

    def warn_stalled(self, threshold: float) -> None:
        """Emit the feature's stall WARNING: a non-parallel handler is still holding its guard."""
        LOGGER.warning(
            "Handler '%s' has held its %s execution-mode guard for over %.0fs and is still running",
            self.handler_short_name,
            self.mode.value,
            threshold,
        )

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

    async def release_guard(self) -> None:
        """Release the execution-mode guard: cancel the in-flight task and drop queued factories.

        Called when a listener is cancelled or replaced so no event/listener/app references leak.
        ``parallel`` listeners hold no guard state, so this is a cheap no-op for them.

        Queued triggers still parked in the guard's deque never spawn a child once released, so
        their outer dispatch tasks are parked on ``done`` futures that nothing else will resolve.
        ``drain_pending_done`` resolves every remaining one so those tasks unwind and
        ``_dispatch_pending`` settles.
        """
        await self.guard.release()
        drain_pending_done(self.pending_done)

    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.

task_bucket: TaskBucket instance-attribute

TaskBucket used to spawn the cancellable child handler task for non-parallel modes.

guard: ExecutionModeGuard instance-attribute

Per-listener overlap state machine. Owns the running-task reference and live counters.

mode: ExecutionMode instance-attribute

Resolved overlap mode. Copied from options so dispatch can branch without reading the guard's private state.

handler_short_name: str instance-attribute

Short handler name, captured for the stall WARNING and debug logs.

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.

backpressure_dropped: int = 0 class-attribute instance-attribute

Count of events dropped at the dispatch acquire gate due to DROP_NEWEST backpressure. Incremented only in BusService.dispatch — one writer, on the event loop, no await between the locked() check and the increment. Live-only; resets on restart, never persisted.

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

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

pending_done: set[asyncio.Future[None]] = field(default_factory=set, init=False) class-attribute instance-attribute

Unresolved per-invocation completion futures for non-parallel modes.

Each run_with_mode call (single/restart/queued) parks its outer dispatch task on a future that resolves when the handler actually runs (or is dropped/released). A queued trigger accepted into the deque has no live child until drain time, so its future would hang forever if the listener is released first. release_guard resolves every remaining future here so those outer dispatch tasks unwind and _dispatch_pending settles.

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,
        task_bucket=task_bucket,
        guard=ExecutionModeGuard(options.mode),
        mode=options.mode,
        handler_short_name=callable_short_name(handler),
        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 the once-guard, rate limiter, and overlap mode around the given invoke function.

Order: once-guard → rate limiter (whether to start) → mode guard (overlap of started invocations). BusService builds invoke_fn (tracked telemetry); HandlerInvoker wraps it.

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

Mode guard: parallel is a pass-through that awaits invoke_fn inline (byte-for-byte today's behavior, no child task). For single/restart/queued the guard receives a run-and-track callable that spawns a fresh child task through task_bucket; this method then awaits that child so the outer dispatch task stays pending and remains counted by _dispatch_pending. A restart cancellation of the child surfaces as CancelledError inside the child only — it is swallowed here so the outer dispatch task does not crash.

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

async def dispatch(self, invoke_fn: Callable[[], Awaitable[None]]) -> None:
    """Apply the once-guard, rate limiter, and overlap mode around the given invoke function.

    Order: once-guard → rate limiter (whether to start) → mode guard (overlap of started
    invocations). BusService builds ``invoke_fn`` (tracked telemetry); HandlerInvoker wraps it.

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

    Mode guard: ``parallel`` is a pass-through that awaits ``invoke_fn`` inline (byte-for-byte
    today's behavior, no child task). For ``single``/``restart``/``queued`` the guard receives a
    run-and-track callable that spawns a fresh child task through ``task_bucket``; this method
    then awaits that child so the outer dispatch task stays pending and remains counted by
    ``_dispatch_pending``. A ``restart`` cancellation of the child surfaces as ``CancelledError``
    inside the child only — it is swallowed here so the outer dispatch task does not crash.
    """
    if self.once and self.fired:
        return
    if self.once:
        self.mark_fired()

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

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

Apply the overlap mode guard to a single started invocation.

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

async def run_with_mode(self, invoke_fn: Callable[[], Awaitable[None]]) -> None:
    """Apply the overlap mode guard to a single started invocation."""
    if self.mode is ExecutionMode.PARALLEL:
        await invoke_fn()
        return

    await run_through_guard(
        guard=self.guard,
        spawn=lambda coro, *, name: self.task_bucket.spawn(coro, name=name),
        pending_done=self.pending_done,
        invoke=invoke_fn,
        warn=self.warn_stalled,
        spawn_name="bus:mode_invocation",
        threshold=STALL_THRESHOLD_SECONDS,
    )

warn_stalled(threshold: float) -> None

Emit the feature's stall WARNING: a non-parallel handler is still holding its guard.

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

def warn_stalled(self, threshold: float) -> None:
    """Emit the feature's stall WARNING: a non-parallel handler is still holding its guard."""
    LOGGER.warning(
        "Handler '%s' has held its %s execution-mode guard for over %.0fs and is still running",
        self.handler_short_name,
        self.mode.value,
        threshold,
    )

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()

release_guard() -> None async

Release the execution-mode guard: cancel the in-flight task and drop queued factories.

Called when a listener is cancelled or replaced so no event/listener/app references leak. parallel listeners hold no guard state, so this is a cheap no-op for them.

Queued triggers still parked in the guard's deque never spawn a child once released, so their outer dispatch tasks are parked on done futures that nothing else will resolve. drain_pending_done resolves every remaining one so those tasks unwind and _dispatch_pending settles.

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

async def release_guard(self) -> None:
    """Release the execution-mode guard: cancel the in-flight task and drop queued factories.

    Called when a listener is cancelled or replaced so no event/listener/app references leak.
    ``parallel`` listeners hold no guard state, so this is a cheap no-op for them.

    Queued triggers still parked in the guard's deque never spawn a child once released, so
    their outer dispatch tasks are parked on ``done`` futures that nothing else will resolve.
    ``drain_pending_done`` resolves every remaining one so those tasks unwind and
    ``_dispatch_pending`` settles.
    """
    await self.guard.release()
    drain_pending_done(self.pending_done)

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)

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,
        normalize_cancel_event: "Callable[[Event[Any]], Event[Any]] | 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,
            normalize_cancel_event=normalize_cancel_event,
        )

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, normalize_cancel_event: Callable[[Event[Any]], Event[Any]] | 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,
    normalize_cancel_event: "Callable[[Event[Any]], Event[Any]] | 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,
        normalize_cancel_event=normalize_cancel_event,
    )

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:
            self.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 and calls invoker.mark_fired(). For once=True listeners, mark_fired()
        prevents the handler from starting on any in-flight dispatch task that has not yet checked
        the once-guard. For once=False listeners, mark_fired() has no effect on dispatch (the
        once-guard is not consulted); protection comes from release_guard() cancelling the in-flight
        child task — spawned asynchronously, so there is a small window before it takes effect.
        Also cancels the rate limiter and duration timer, and releases the execution-mode guard
        (cancelling any in-flight handler task and dropping queued factories) so no event/listener
        references leak.

        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()
        # release_guard is async (it awaits the cancelled task's settling under a lock); cancel()
        # is sync, so spawn the release on the same bucket that runs handler tasks. For ``parallel``
        # listeners this is a cheap no-op; for the others it drops the in-flight task and queue.
        self.invoker.task_bucket.spawn(self.invoker.release_guard(), name="bus:release_guard")

    def config_matches(self, other: "Listener") -> bool:
        """Check whether two listeners represent the same logical configuration.

        Compares handler callable, filter predicate, timing options (once, debounce,
        throttle, timeout, timeout_disabled, priority), the execution mode, handler kwargs,
        per-registration error handler (by identity), and duration configuration scalars.

        Does not compare runtime state: listener_id, db_id, _cancelled, or the
        attached DurationTimer. Lambda/closure predicates and callable conditions
        compare by identity — two fresh lambdas with identical bodies will report
        drift. Use non-lambda predicates or if_exists='replace' to avoid this.
        """
        return (
            self.invoker.orig_handler == other.invoker.orig_handler
            and self.predicate == other.predicate
            and self.options.once == other.options.once
            and self.options.debounce == other.options.debounce
            and self.options.throttle == other.options.throttle
            and self.options.timeout == other.options.timeout
            and self.options.timeout_disabled == other.options.timeout_disabled
            and self.options.priority == other.options.priority
            and self.options.mode == other.options.mode
            and self.options.backpressure == other.options.backpressure
            and self.invoker.kwargs == other.invoker.kwargs
            and self.invoker.error_handler is other.invoker.error_handler
            and _duration_configs_match(self.duration_config, other.duration_config)
        )

    def diff_fields(self, other: "Listener") -> list[str]:
        """Return configuration field names that differ between two listeners.

        Compares the same fields as config_matches(). Returns a stable-ordered list
        of field names (e.g. 'handler', 'predicate', 'once', 'debounce', ...) for use
        in drift error messages.
        """
        changed: list[str] = []
        if self.invoker.orig_handler != other.invoker.orig_handler:
            changed.append("handler")
        if self.predicate != other.predicate:
            changed.append("predicate")
        if self.options.once != other.options.once:
            changed.append("once")
        if self.options.debounce != other.options.debounce:
            changed.append("debounce")
        if self.options.throttle != other.options.throttle:
            changed.append("throttle")
        if self.options.timeout != other.options.timeout:
            changed.append("timeout")
        if self.options.timeout_disabled != other.options.timeout_disabled:
            changed.append("timeout_disabled")
        if self.options.priority != other.options.priority:
            changed.append("priority")
        if self.options.mode != other.options.mode:
            changed.append("mode")
        if self.options.backpressure != other.options.backpressure:
            changed.append("backpressure")
        if self.invoker.kwargs != other.invoker.kwargs:
            changed.append("kwargs")
        if self.invoker.error_handler is not other.invoker.error_handler:
            changed.append("error_handler")
        if not _duration_configs_match(self.duration_config, other.duration_config):
            changed.append("duration_config")
        return changed

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

        Raises if the predicate raises — the caller (BusService.dispatch) handles
        isolation, telemetry recording, and error-handler routing.
        """
        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: WhereClause = 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",
        )

        # Cancel-listeners are source_tier='framework' but bypass _on_internal's tier-aware
        # resolution, so set parallel explicitly to match the framework-tier default. They fire at
        # most once per timer, so parallel is the safe internal default.
        options = ListenerOptions(mode=ExecutionMode.PARALLEL)

        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:
        self.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 and calls invoker.mark_fired(). For once=True listeners, mark_fired() prevents the handler from starting on any in-flight dispatch task that has not yet checked the once-guard. For once=False listeners, mark_fired() has no effect on dispatch (the once-guard is not consulted); protection comes from release_guard() cancelling the in-flight child task — spawned asynchronously, so there is a small window before it takes effect. Also cancels the rate limiter and duration timer, and releases the execution-mode guard (cancelling any in-flight handler task and dropping queued factories) so no event/listener references leak.

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 and calls invoker.mark_fired(). For once=True listeners, mark_fired()
    prevents the handler from starting on any in-flight dispatch task that has not yet checked
    the once-guard. For once=False listeners, mark_fired() has no effect on dispatch (the
    once-guard is not consulted); protection comes from release_guard() cancelling the in-flight
    child task — spawned asynchronously, so there is a small window before it takes effect.
    Also cancels the rate limiter and duration timer, and releases the execution-mode guard
    (cancelling any in-flight handler task and dropping queued factories) so no event/listener
    references leak.

    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()
    # release_guard is async (it awaits the cancelled task's settling under a lock); cancel()
    # is sync, so spawn the release on the same bucket that runs handler tasks. For ``parallel``
    # listeners this is a cheap no-op; for the others it drops the in-flight task and queue.
    self.invoker.task_bucket.spawn(self.invoker.release_guard(), name="bus:release_guard")

config_matches(other: Listener) -> bool

Check whether two listeners represent the same logical configuration.

Compares handler callable, filter predicate, timing options (once, debounce, throttle, timeout, timeout_disabled, priority), the execution mode, handler kwargs, per-registration error handler (by identity), and duration configuration scalars.

Does not compare runtime state: listener_id, db_id, _cancelled, or the attached DurationTimer. Lambda/closure predicates and callable conditions compare by identity — two fresh lambdas with identical bodies will report drift. Use non-lambda predicates or if_exists='replace' to avoid this.

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

def config_matches(self, other: "Listener") -> bool:
    """Check whether two listeners represent the same logical configuration.

    Compares handler callable, filter predicate, timing options (once, debounce,
    throttle, timeout, timeout_disabled, priority), the execution mode, handler kwargs,
    per-registration error handler (by identity), and duration configuration scalars.

    Does not compare runtime state: listener_id, db_id, _cancelled, or the
    attached DurationTimer. Lambda/closure predicates and callable conditions
    compare by identity — two fresh lambdas with identical bodies will report
    drift. Use non-lambda predicates or if_exists='replace' to avoid this.
    """
    return (
        self.invoker.orig_handler == other.invoker.orig_handler
        and self.predicate == other.predicate
        and self.options.once == other.options.once
        and self.options.debounce == other.options.debounce
        and self.options.throttle == other.options.throttle
        and self.options.timeout == other.options.timeout
        and self.options.timeout_disabled == other.options.timeout_disabled
        and self.options.priority == other.options.priority
        and self.options.mode == other.options.mode
        and self.options.backpressure == other.options.backpressure
        and self.invoker.kwargs == other.invoker.kwargs
        and self.invoker.error_handler is other.invoker.error_handler
        and _duration_configs_match(self.duration_config, other.duration_config)
    )

diff_fields(other: Listener) -> list[str]

Return configuration field names that differ between two listeners.

Compares the same fields as config_matches(). Returns a stable-ordered list of field names (e.g. 'handler', 'predicate', 'once', 'debounce', ...) for use in drift error messages.

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

def diff_fields(self, other: "Listener") -> list[str]:
    """Return configuration field names that differ between two listeners.

    Compares the same fields as config_matches(). Returns a stable-ordered list
    of field names (e.g. 'handler', 'predicate', 'once', 'debounce', ...) for use
    in drift error messages.
    """
    changed: list[str] = []
    if self.invoker.orig_handler != other.invoker.orig_handler:
        changed.append("handler")
    if self.predicate != other.predicate:
        changed.append("predicate")
    if self.options.once != other.options.once:
        changed.append("once")
    if self.options.debounce != other.options.debounce:
        changed.append("debounce")
    if self.options.throttle != other.options.throttle:
        changed.append("throttle")
    if self.options.timeout != other.options.timeout:
        changed.append("timeout")
    if self.options.timeout_disabled != other.options.timeout_disabled:
        changed.append("timeout_disabled")
    if self.options.priority != other.options.priority:
        changed.append("priority")
    if self.options.mode != other.options.mode:
        changed.append("mode")
    if self.options.backpressure != other.options.backpressure:
        changed.append("backpressure")
    if self.invoker.kwargs != other.invoker.kwargs:
        changed.append("kwargs")
    if self.invoker.error_handler is not other.invoker.error_handler:
        changed.append("error_handler")
    if not _duration_configs_match(self.duration_config, other.duration_config):
        changed.append("duration_config")
    return changed

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

Check if the event matches the listener's predicate.

Raises if the predicate raises — the caller (BusService.dispatch) handles isolation, telemetry recording, and error-handler routing.

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

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

    Raises if the predicate raises — the caller (BusService.dispatch) handles
    isolation, telemetry recording, and error-handler routing.
    """
    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: WhereClause = 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: WhereClause = 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",
    )

    # Cancel-listeners are source_tier='framework' but bypass _on_internal's tier-aware
    # resolution, so set parallel explicitly to match the framework-tier default. They fire at
    # most once per timer, so parallel is the safe internal default.
    options = ListenerOptions(mode=ExecutionMode.PARALLEL)

    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,
    )

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()

make_async_handler(fn: HandlerType, task_bucket: TaskBucket) -> AsyncHandlerType

Wrap a function to ensure it is always called as an async handler.

If the function is already an async function, it will be called directly. If it is a regular function, it will be run in an executor to avoid blocking the event loop.

Parameters:

Name	Type	Description	Default
fn	HandlerType	The function to adapt.	required
task_bucket	TaskBucket	TaskBucket used to create the async adapter (runs sync handlers in executor).	required

Returns:

Type	Description
AsyncHandlerType	An async handler that wraps the original function.

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

def make_async_handler(fn: "HandlerType", task_bucket: "TaskBucket") -> "AsyncHandlerType":
    """Wrap a function to ensure it is always called as an async handler.

    If the function is already an async function, it will be called directly.
    If it is a regular function, it will be run in an executor to avoid blocking the event loop.

    Args:
        fn: The function to adapt.
        task_bucket: TaskBucket used to create the async adapter (runs sync handlers in executor).

    Returns:
        An async handler that wraps the original function.
    """
    return cast("AsyncHandlerType", task_bucket.make_async_adapter(fn))