Base

FinalMeta

Bases: type

Disallow overriding methods marked @final in any ancestor.

Source code in src/hassette/resources/base.py

class FinalMeta(type):
    """Disallow overriding methods marked @final in any ancestor."""

    LOADED_CLASSES: ClassVar[set[str]] = set()

    def __init__(cls, name, bases, ns, **kw):
        super().__init__(name, bases, ns, **kw)
        subclass_name = f"{cls.__module__}.{cls.__qualname__}"
        if subclass_name in FinalMeta.LOADED_CLASSES:
            return

        FinalMeta.LOADED_CLASSES.add(subclass_name)

        # Collect all methods marked as final from the MRO (excluding object and cls itself)
        finals: dict[str, type] = {}
        for ancestor in cls.__mro__[1:]:
            if ancestor is object:
                continue
            for attr, obj in ancestor.__dict__.items():
                if getattr(obj, "__final__", False):
                    finals.setdefault(attr, ancestor)

        for method_name, origin in finals.items():
            if method_name in ns:
                new_obj = ns[method_name]
                old_obj = origin.__dict__.get(method_name)
                if new_obj is old_obj:
                    continue

                origin_name = f"{origin.__qualname__}"
                subclass_name = f"{cls.__module__}.{cls.__qualname__}"
                suggested_alt = f"on_{method_name}" if not method_name.startswith("on_") else method_name

                loc = None
                code = getattr(new_obj, "__code__", None)
                if code is not None:
                    loc = f"{code.co_filename}:{code.co_firstlineno}"

                raise CannotOverrideFinalError(method_name, origin_name, subclass_name, suggested_alt, loc)

Resource

Bases: LifecycleMixin

Base class for resources in the Hassette framework.

Source code in src/hassette/resources/base.py

class Resource(LifecycleMixin, metaclass=FinalMeta):
    """Base class for resources in the Hassette framework."""

    _unique_name: str
    """Unique name for the instance."""

    role: ClassVar[ResourceRole] = ResourceRole.RESOURCE
    """Role of the resource, e.g. 'App', 'Service', etc."""

    depends_on: ClassVar[list[type["Resource"]]] = []
    """Resource types that must be ready before this resource initializes."""

    source_tier: ClassVar[SourceTier] = "framework"
    """Telemetry classification inherited by Bus/Scheduler children for DB registration.

    Defaults to ``'framework'`` for all Resources. User-facing app classes (``App``,
    ``AppSync``) override to ``'app'``. Do not set ``source_tier = 'app'`` on framework
    components — their Bus/Scheduler children inherit this value and it determines
    cleanup, reconciliation, and UI display behavior.
    """

    index: int = 0
    """Instance index. Apps override with their manifest-assigned index."""

    task_bucket: "TaskBucket"
    """Task bucket for managing tasks owned by this instance."""

    is_task_bucket: ClassVar[bool] = False
    """True on TaskBucket (and any subclass that keeps it True, since it is inherited); used in
    __init__ so a Resource that is its own task bucket skips the factory, avoiding a circular import."""

    _default_task_bucket_factory: ClassVar["Callable[[Hassette, Resource], TaskBucket] | None"] = None
    """Factory registered by hassette.task_bucket at import time; raises if unset."""

    parent: "Resource | None" = None
    """Reference to the parent resource, if any."""

    children: list["Resource"]
    """List of child resources."""

    logger: Logger
    """Logger for the instance."""

    unique_id: str
    """Unique identifier for the instance."""

    class_name: typing.ClassVar[str]
    """Name of the class, set on subclassing."""

    hassette: "Hassette"
    """Reference to the Hassette instance."""

    def __init_subclass__(cls) -> None:
        cls.class_name = cls.__name__
        if "depends_on" not in cls.__dict__:
            cls.depends_on = list(cls.depends_on)

    def __init__(
        self, hassette: "Hassette", task_bucket: "TaskBucket | None" = None, parent: "Resource | None" = None
    ) -> None:
        super().__init__()

        self.unique_id = uuid.uuid4().hex[:8]

        self.hassette = hassette
        self.parent = parent
        self.children = []

        self._setup_logger()

        if self.is_task_bucket:
            # TaskBucket is special: it is its own task bucket. pyright can't narrow `self` to
            # TaskBucket through the `is_task_bucket: ClassVar[bool]` guard, so the assignment of
            # `self` to the TaskBucket-typed attribute needs the suppression.
            self.task_bucket = self  # pyright: ignore[reportAttributeAccessIssue]
        else:
            if task_bucket is not None:
                self.task_bucket = task_bucket
            else:
                factory = Resource._default_task_bucket_factory
                if factory is None:
                    raise RuntimeError(
                        f"Cannot construct {type(self).__name__}: no TaskBucket factory is registered. "
                        "Ensure hassette.task_bucket is imported before constructing any Resource."
                    )
                self.task_bucket = factory(self.hassette, self)

    def _get_logger_name(self) -> str:
        # Asks the parent for its name rather than reading parent.logger, which doesn't exist yet when
        # Hassette builds its own TaskBucket child.
        # A root resource is its own parent; recursing into it would never end. (Hassette overrides this method.)
        if self.parent is None or self.parent is self:
            return f"hassette.{self.unique_name}"
        return f"{self.parent._get_logger_name()}.{self.class_name}"

    def _setup_logger(self) -> None:
        self.logger = getLogger(self._get_logger_name())
        # Only an app's own logger is app-authored; its children and every framework resource write
        # framework lines.
        author: LogAuthor = "app" if self.role is ResourceRole.APP else "framework"
        apply_log_level(self.logger, self.resolve_log_level(author, self.config_log_level), self.unique_name)
        set_logger_identity(self.logger, self.owner_log_identity(author))

    def resolve_log_level(self, author: LogAuthor, default_level: LogLevel) -> LogLevel:
        """Level for a logger this resource builds: the owning app's override, else ``default_level``.

        A framework-authored logger takes the owning app's ``framework_log_level`` when it is set. An
        app-authored logger always keeps ``default_level``, since the app's own ``log_level`` already feeds
        `App.config_log_level`.
        """
        if author == "app":
            return default_level
        return self.owner_framework_log_level() or default_level

    def owner_framework_log_level(self) -> LogLevel | None:
        """The owning app's ``framework_log_level``, or None when no app owns this resource or it is unset.

        Found the same way as `owner_log_identity`: App answers for itself, every other resource asks its
        parent.
        """
        if self.parent is None or self.parent is self:
            return None
        return self.parent.owner_framework_log_level()

    @property
    def lifecycle_logger(self) -> Logger:
        """Logger generic lifecycle code writes through about this resource.

        The resource's own logger. App overrides it with a framework-authored logger the app owns,
        so lifecycle lines about an app aren't attributed to the app's code.
        """
        return self.logger

    def owner_log_identity(self, author: LogAuthor) -> LogIdentity:
        """Identity for a logger this resource builds: the given ``author`` plus the owning app instance.

        The owner is the nearest App at or above this resource. App overrides this to name itself;
        every other resource asks its parent, so an App's children and their descendants carry the
        app, and framework resources reach the root and carry no owner.
        """
        # A root resource is its own parent (see _get_logger_name).
        if self.parent is None or self.parent is self:
            return LogIdentity(author=author)
        return self.parent.owner_log_identity(author)

    def __repr__(self) -> str:
        return f"<{type(self).__name__} unique_name={self.unique_name}>"

    @property
    def unique_name(self) -> str:
        """Get the unique name of the instance."""
        if not hasattr(self, "_unique_name") or not self._unique_name:
            if self.parent:
                self._unique_name = f"{self.parent.unique_name}.{self.class_name}"
            else:
                self._unique_name = f"{self.class_name}.{self.unique_id}"

        return self._unique_name

    @property
    def app_key(self) -> str:
        """Identity key for telemetry. App overrides with its manifest key."""
        return f"{FRAMEWORK_APP_KEY_PREFIX}{self.class_name}"

    @property
    def instance_name(self) -> str | None:
        """Owning app instance's name for telemetry, or None for framework resources.

        Resolved from the resource's ``app_config`` when present (App subclasses);
        plain framework resources have no app config and return None.
        """
        return getattr(getattr(self, "app_config", None), "instance_name", None)

    @property
    def owner_id(self) -> str:
        """Unique name of the owning resource: the parent's ``unique_name``, else this resource's own."""
        # nearest App's unique_name, else Hassette's unique_name
        if self.parent:
            return self.parent.unique_name
        return self.unique_name

    @property
    def config_log_level(self) -> LogLevel:
        """Return the log level from the config for this resource.

        Subclasses that map to a per-service ``LoggingConfig`` field override this with a
        one-line property returning that field. That repetition is deliberate and stays:
        the alternative — a class attribute naming the field, read here via ``getattr`` —
        trades a checked attribute access for a string, and the pattern spans 30 files,
        three of which are generated sync facades. ``tools/check_duplicate_code.py`` reports
        the overrides as clones because its matched region also swallows the unrelated
        statement above each one, so they cannot be annotated with ``dup-ignore`` markers
        without mislabeling that statement (see #1570).
        """
        return self.hassette.config.logging.log_level

    @classmethod
    def _validate_parent(cls, parent: "Resource") -> None:
        """Raise when ``parent`` may not own this class; ``add_child()`` calls it. Any parent owns a plain resource."""

    def _optional_serve_task(self) -> "asyncio.Task[None] | None":
        """The serve task of an optional service; None for every other resource.

        A startup wave treats a member whose optional serve task has ended as settled (spec 134 D7).
        """
        return None

    def add_child(self, child_class: type[_ResourceT], **kwargs: Any) -> _ResourceT:
        """Create and add a child resource to this resource.

        Args:
            child_class: The class of the child resource to create.
            **kwargs: Keyword arguments to pass to the child resource's constructor.

        Returns:
            The created child resource.
        """
        if "parent" in kwargs:
            raise ValueError("Cannot specify 'parent' argument when adding a child resource; it is set automatically.")
        child_class._validate_parent(self)

        inst = child_class(hassette=self.hassette, parent=self, **kwargs)
        self.children.append(inst)
        return inst

    async def _auto_wait_dependencies(self) -> None:
        """Wait for all declared depends_on types to become ready before lifecycle hooks fire.

        Early-returns when:
        - ``depends_on`` is empty (no declared deps)
        - ``hassette.should_skip_dependency_check()`` returns True (test harness bypass)

        Raises:
            RuntimeError: If no matching child is found for a declared dep type, or if
                ``hassette.wait_for_ready`` returns False without a concurrent shutdown signal.

        On shutdown during wait, calls ``mark_not_ready()`` and returns without raising.
        """
        if not self.depends_on:
            return
        if self.hassette.should_skip_dependency_check():
            return

        # App-level depends_on is not yet supported (#581).
        if self.role == ResourceRole.APP:
            raise RuntimeError(
                f"{self.class_name} declares depends_on but App-level depends_on "
                f"is not yet supported. See https://github.com/NodeJSmith/hassette/issues/581"
            )

        # Deduplicates by instance identity (id), not by type — necessary because
        # a single child instance may satisfy multiple dep_type entries (e.g.,
        # depends_on = [Service, DatabaseService] where DatabaseService matches both).
        seen: set[int] = set()
        deps: list[Resource] = []
        for dep_type in self.depends_on:
            matches = [child for child in self.hassette.children if isinstance(child, dep_type)]
            if not matches:
                raise RuntimeError(
                    f"{self.class_name} declares depends_on=[{dep_type.__name__}] "
                    f"but no matching child found in Hassette"
                )
            for match in matches:
                if id(match) not in seen:
                    seen.add(id(match))
                    deps.append(match)

        dep_names = ", ".join(dep.class_name for dep in deps)
        self.lifecycle_logger.debug("Waiting for dependencies: [%s]", dep_names)

        ready = await self.hassette.wait_for_ready(deps)
        if not ready:
            if self.hassette.shutdown_event.is_set():
                mark_not_ready(self, "shutdown during dependency wait")
                return
            status_report = ", ".join(f"{dep.class_name}({dep.status.value})" for dep in deps)
            raise RuntimeError(f"{self.class_name} timed out waiting for dependencies: {status_report}")

        self.lifecycle_logger.debug("Dependencies satisfied: [%s]", dep_names)

    def _force_terminal(self) -> None:
        """Recursively force this resource and all descendants to STOPPED terminal state.

        Cancels tasks for resources that were never given a shutdown signal (grandchildren).
        Service overrides this to also cancel _serve_task.

        Stores an ``INCOMPLETE`` shutdown outcome on the resource before cancelling any work, so
        restart is refused even if a caller inspects the outcome before this method's cancellation and
        terminal-status side effects finish. Leaves an already completed outcome unchanged — a resource
        that already proved ``is_restart_safe`` (or already recorded an unsafe outcome) is not
        retroactively altered.

        Note: this does NOT call on_shutdown() hooks, so bus subscriptions and scheduler
        jobs owned by force-terminated resources are not cleaned up. This is intentional —
        calling hooks risks re-entrancy with the child's own finally block. Stale
        subscriptions may remain active against STOPPED resources; this is an accepted
        gap because force-terminal is nearly always followed by process exit.

        Seals the TaskBucket and takes its synchronous pending-name snapshot before recording
        the outcome, so a force-terminal call from a deadline owner walking away (the root's exit
        stages, ``AppLifecycleService``'s entry deadline) while ``_shutdown_body()`` never reached its
        own TaskBucket stage -- see ``_run_task_bucket_shutdown_stage()`` -- still records TaskBucket's
        final pending names even though the interrupted body's own stage never returned.

        The "leave an already-completed outcome unchanged" guard below only skips re-recording
        that outcome -- it does NOT skip the recursion into children. A resource whose own outcome
        was already stored (e.g. a clean teardown, or an earlier force-terminal call) can still
        have unresponsive descendants of its own; a deadline owner force-terminates every
        unfinished child of its stage specifically to reach those descendants, so returning early
        here would leave a hung grandchild's tasks alive after shutdown has already declared the
        tree terminal.
        """
        if self._shutdown_outcome is None:
            self.task_bucket.seal()
            self._shutdown_outcome = incomplete_outcome(
                f"{self.unique_name} was force-terminated", self.task_bucket.pending_task_names()
            )
        cancel(self)
        # Cancel an active shutdown coordinator too -- but never the currently running task
        # (self-cancellation mid-synchronous-execution is a footgun and would not take effect
        # until the next suspension point anyway). This is the cross-resource case: a parent's
        # own shutdown body force-terminating an unresponsive child whose shutdown coordinator
        # is still pending elsewhere. Because the outcome above is stored first, the child's
        # coordinator (`_run_shutdown_coordinator`) treats the resulting cancellation as a
        # restart-unsafe return rather than letting `CancelledError` reach its joined callers.
        current_task = asyncio.current_task()
        if (
            self._shutdown_task is not None
            and not self._shutdown_task.done()
            and self._shutdown_task is not current_task
        ):
            self._shutdown_task.cancel()
        self.task_bucket.cancel_all_sync()
        self._status = ResourceStatus.STOPPED  # bypass setter to skip validation
        mark_not_ready(self, "shutdown timed out")
        for child in self.children:
            child._force_terminal()

    async def _shutdown_children(self) -> ShutdownOutcome:
        """Propagate shutdown to children and merge their shutdown outcomes.

        Children shut down concurrently, in reverse insertion order
        (``ordered_children_for_shutdown()``), as a single ``shutdown_batch()`` call — see that
        function for how each child's outcome merges in. The batch has no
        bound of its own: each child inherits this resource's shutdown deadline.
        """
        children = ordered_children_for_shutdown(self)
        if not children:
            return ShutdownOutcome()

        return await shutdown_batch(self, children, deadline=self._shutdown_deadline)

    async def _run_task_bucket_shutdown_stage(self) -> ShutdownOutcome:
        """Seal the TaskBucket, cancel tracked work, and name the tasks still pending.

        First-class shutdown stage: sealing, cancellation, and the final synchronous pending-name
        snapshot all happen here, before subclass ``cleanup()`` runs (design: "Sealing,
        cancellation, and final TaskBucket inspection form a first-class shutdown stage before
        subclass cleanup"). Kept separate from ``cleanup()`` so an enclosing whole-body timeout or
        force-terminal call can still inspect the sealed bucket directly (see
        ``_force_terminal()``) even if this stage itself is interrupted before returning.

        Bounded by ``task_cancellation_timeout_seconds`` clamped to the attempt's shutdown deadline.
        """
        self.task_bucket.seal()
        cancel_budget = clamp_to_shutdown_deadline(
            self, self.hassette.config.lifecycle.task_cancellation_timeout_seconds
        )
        self.lifecycle_logger.debug(
            "%s: entering task-bucket cancel with %.2fs bound",
            self.unique_name,
            cancel_budget,
        )
        cancel_start = asyncio.get_running_loop().time()
        await self.task_bucket.cancel_all(timeout=cancel_budget)
        self.lifecycle_logger.debug(
            "%s: task-bucket cancel stage completed in %.2fs",
            self.unique_name,
            elapsed_since(cancel_start),
        )
        return ShutdownOutcome(still_running=self.task_bucket.pending_task_names())

    async def _run_post_hook_shutdown_stage(self) -> ShutdownOutcome:
        """Run the post-hook shutdown stage shared by every ``_shutdown_body()`` implementation.

        Covers cleanup, child propagation, and terminal event emission for Resource, Service, and Hassette.

        Runs after the class-specific shutdown hooks have already executed. Returns an outcome
        instead of setting flags directly — the shutdown coordinator (``coordinate_shutdown()``
        in ``hassette.resources.lifecycle``) is the sole owner of the stored ``ShutdownOutcome``.
        """
        outcomes: list[ShutdownOutcome] = [await self._run_task_bucket_shutdown_stage()]

        cleanup_timeout = clamp_to_shutdown_deadline(self, CLEANUP_SECONDS)
        self.lifecycle_logger.debug(
            "%s: entering cleanup() with %.2fs bound",
            self.unique_name,
            cleanup_timeout,
        )
        cleanup_start = asyncio.get_running_loop().time()
        try:
            async with asyncio.timeout(cleanup_timeout):
                await self.cleanup()
        except TimeoutError:
            self.lifecycle_logger.warning("cleanup() timed out after %.2fs for %s", cleanup_timeout, self.unique_name)
            reason = f"{self.unique_name}: cleanup() timed out after {cleanup_timeout:.2f}s"
            # FAILED, not TIMED_OUT: an app stop reads TIMED_OUT as "a shutdown hook timed out" and fails the stop.
            outcomes.append(ShutdownOutcome(ShutdownResult.FAILED, reason=reason))
        except Exception as exc:
            self.lifecycle_logger.exception("Error during cleanup: %s %s", type(exc).__name__, exc)
            reason = f"{self.unique_name}: cleanup() raised {type(exc).__name__}"
            outcomes.append(ShutdownOutcome(ShutdownResult.FAILED, reason=reason))
        self.lifecycle_logger.debug(
            "%s: cleanup() stage completed in %.2fs",
            self.unique_name,
            elapsed_since(cleanup_start),
        )

        children_start = asyncio.get_running_loop().time()
        children_outcome = await self._shutdown_children()
        self.lifecycle_logger.debug(
            "%s: child shutdown propagation completed in %.2fs",
            self.unique_name,
            elapsed_since(children_start),
        )
        outcomes.append(children_outcome)

        if children_outcome.is_restart_safe:
            await self._on_children_stopped()

        # Runs with closed streams too: the transition still happens and only its event is dropped (spec 134 D9).
        try:
            await handle_stop(self)
        except Exception as exc:
            self.lifecycle_logger.exception("Error during stopping %s %s", type(exc).__name__, exc)

        return merge_outcomes(*outcomes)

    async def _on_children_stopped(self) -> None:
        """Called after children shut down, before this resource's STOPPED event.

        Runs when the children's merged outcome is restart-safe, meaning no child's work may still
        be running. A child hook that raised or timed out and has ended doesn't block it. Skipped when a
        child left work pending or its evidence is incomplete (it was force-terminated, or its
        ``shutdown()`` raised); the caller handles fallback teardown then, e.g. Hassette's finally
        block calls close_streams().

        Override to run logic that must happen after children are shut down but
        before the parent emits its own STOPPED event. Default is a no-op.
        Overrides MUST call ``await super()._on_children_stopped()``.

        Note: _shutdown_body() is intentionally not @final — this hook exists so
        subclasses do NOT need to override _shutdown_body() for post-children behavior.
        """

    @final
    async def initialize(self) -> None:
        """Coordinator front door: joins or creates the resource's one owned initialization attempt.

        This method itself never runs class-specific initialization work — see
        ``_initialize_body()``, which every initialization attempt (direct calls and ``start()``'s
        spawned joiner) ultimately runs exactly once per attempt. See
        ``coordinate_initialize()`` in ``hassette.resources.lifecycle`` for the full admission
        sequence (re-entry rejection, joining an active shutdown, restart refusal, task
        ownership).
        """
        await coordinate_initialize(self)

    async def _initialize_body(self) -> None:
        """Class-specific initialization work, run exactly once per coordinated attempt.

        NOTE: keep child propagation in sync with Service._initialize_body().
        NOTE: _auto_wait_dependencies() runs before hooks — keep in sync with Service.
        """
        self.lifecycle_logger.debug("Initializing %s: %s", self.role, self.unique_name)
        await handle_starting(self)

        # Only Exception: a CancelledError here comes from cancel() or shutdown(), which own that
        # outcome, so the attempt is intentionally not marked FAILED or logged as a failure.
        try:
            await self._auto_wait_dependencies()
        except Exception as exc:
            await handle_failed(self, exc)
            raise
        if self.hassette.shutdown_event.is_set():
            mark_not_ready(self, "shutdown requested during dependency wait")
            return
        await run_hooks(
            self, [self._framework_before_hooks, self.before_initialize, self.on_initialize, self.after_initialize]
        )
        for child in self.children:
            if child.status not in (ResourceStatus.STARTING, ResourceStatus.RUNNING):
                await child.initialize()
        await handle_running(self)

    async def _framework_before_hooks(self) -> None:
        """Framework-owned setup that runs before ``before_initialize()``.

        Subclasses inside the framework override this for setup that must happen regardless of
        how a user overrides the lifecycle hooks, so those hooks stay empty and never need a
        ``super()`` call. It runs inside the same ``run_hooks()`` call as the user hooks, so a
        failure here is handled exactly like a hook failure.
        """
        pass

    async def before_initialize(self) -> None:
        """Optional: prepare to accept new work, allocate sockets, queues, temp files, etc."""
        pass

    async def on_initialize(self) -> None:
        """Primary hook: perform your own initialization (sockets, queues, temp files…)."""
        pass

    async def after_initialize(self) -> None:
        """Optional: finalize initialization, signal readiness, etc."""
        pass

    @final
    async def shutdown(self) -> ShutdownOutcome:
        """Coordinator front door: joins or creates the resource's one owned shutdown attempt.

        Returns the attempt's stored ``ShutdownOutcome``.

        This method itself never runs class-specific shutdown work — see ``_shutdown_body()``,
        which the coordinator runs exactly once per attempt. Repeated calls after completion
        return the same stored outcome without rerunning hooks. See ``coordinate_shutdown()`` in
        ``hassette.resources.lifecycle`` for the full sequence (re-entry rejection, initializer
        cancellation/observation, bounded body observation, force-terminal outcome merging).
        """
        return await coordinate_shutdown(self)

    async def _shutdown_body(self) -> ShutdownOutcome:
        """Class-specific shutdown work, run exactly once per coordinated attempt.

        NOTE: keep hook ordering in sync with Service._shutdown_body().

        Skips the resource's own shutdown hooks when its owner set ``_skip_own_shutdown_hooks`` for this
        attempt; the post-hook stage runs either way.
        """
        if self._skip_own_shutdown_hooks:
            self.lifecycle_logger.debug("%s: skipping own shutdown hooks for this attempt", self.unique_name)
            hook_outcome = ShutdownOutcome()
        else:
            hook_errors = await run_hooks(
                self,
                [self.before_shutdown, self.on_shutdown, self.after_shutdown],
                continue_on_error=True,
                shutdown_bound=True,
            )
            hook_outcome = shutdown_hook_outcome(self.unique_name, hook_errors)
        tail_outcome = await self._run_post_hook_shutdown_stage()
        return merge_outcomes(hook_outcome, tail_outcome)

    async def before_shutdown(self) -> None:
        """Optional: stop accepting new work, signal loops to wind down, etc."""
        pass

    async def on_shutdown(self) -> None:
        """Primary hook: release your own stuff (sockets, queues, temp files…)."""
        pass

    async def after_shutdown(self) -> None:
        """Optional: last-chance actions after on_shutdown, before cleanup/STOPPED."""
        pass

    async def _emit_readiness_event(self) -> None:
        """Emit a service_status event reflecting the current readiness state.

        Call this from an async context (e.g., inside ``serve()``) after calling
        ``mark_ready()`` or ``mark_not_ready()`` to propagate mid-operation
        readiness changes to the frontend.

        This method is intended for mid-operation readiness changes while the service
        status is RUNNING. Do not call after handle_failed(), handle_stop(), or
        handle_crash() — those lifecycle methods emit their own status events including
        the current readiness state. Calling this method after a lifecycle transition
        will emit a duplicate event.

        Exceptions are caught internally and logged as warnings — callers do not need
        to wrap with ``suppress(Exception)``.
        """
        try:
            event = create_service_status_event(
                self, self._status, ready=self.is_ready(), ready_phase=self._ready_reason
            )
            await self.hassette.send_event(event)
        except Exception:
            self.lifecycle_logger.warning(
                "%s failed to emit readiness event (ready=%s, phase=%s)",
                self.unique_name,
                self.is_ready(),
                self._ready_reason,
                exc_info=True,
            )

    async def cleanup(self) -> None:
        """Cleanup resources owned by the instance.

        Called during shutdown, after the shutdown coordinator has already cancelled and observed
        any active initialization task, and after ``_run_task_bucket_shutdown_stage()`` has sealed
        and cancelled the TaskBucket's tracked work. Subclass overrides use this for resources they
        own directly (caches, connections, etc.); the base implementation owns nothing.
        """
        self.lifecycle_logger.debug("Cleaned up resources")

role: ResourceRole = ResourceRole.RESOURCE class-attribute

Role of the resource, e.g. 'App', 'Service', etc.

depends_on: list[type[Resource]] = [] class-attribute

Resource types that must be ready before this resource initializes.

source_tier: SourceTier = 'framework' class-attribute

Telemetry classification inherited by Bus/Scheduler children for DB registration.

Defaults to 'framework' for all Resources. User-facing app classes (App, AppSync) override to 'app'. Do not set source_tier = 'app' on framework components — their Bus/Scheduler children inherit this value and it determines cleanup, reconciliation, and UI display behavior.

index: int = 0 class-attribute instance-attribute

Instance index. Apps override with their manifest-assigned index.

task_bucket: TaskBucket instance-attribute

Task bucket for managing tasks owned by this instance.

is_task_bucket: bool = False class-attribute

True on TaskBucket (and any subclass that keeps it True, since it is inherited); used in init so a Resource that is its own task bucket skips the factory, avoiding a circular import.

logger: Logger instance-attribute

Logger for the instance.

class_name: str class-attribute

Name of the class, set on subclassing.

unique_id: str = uuid.uuid4().hex[:8] instance-attribute

Unique identifier for the instance.

hassette: Hassette = hassette instance-attribute

Reference to the Hassette instance.

parent: Resource | None = parent class-attribute instance-attribute

Reference to the parent resource, if any.

children: list[Resource] = [] instance-attribute

List of child resources.

lifecycle_logger: Logger property

Logger generic lifecycle code writes through about this resource.

The resource's own logger. App overrides it with a framework-authored logger the app owns, so lifecycle lines about an app aren't attributed to the app's code.

unique_name: str property

Get the unique name of the instance.

app_key: str property

Identity key for telemetry. App overrides with its manifest key.

instance_name: str | None property

Owning app instance's name for telemetry, or None for framework resources.

Resolved from the resource's app_config when present (App subclasses); plain framework resources have no app config and return None.

owner_id: str property

Unique name of the owning resource: the parent's unique_name, else this resource's own.

config_log_level: LogLevel property

Return the log level from the config for this resource.

Subclasses that map to a per-service LoggingConfig field override this with a one-line property returning that field. That repetition is deliberate and stays: the alternative — a class attribute naming the field, read here via getattr — trades a checked attribute access for a string, and the pattern spans 30 files, three of which are generated sync facades. tools/check_duplicate_code.py reports the overrides as clones because its matched region also swallows the unrelated statement above each one, so they cannot be annotated with dup-ignore markers without mislabeling that statement (see #1570).

resolve_log_level(author: LogAuthor, default_level: LogLevel) -> LogLevel

Level for a logger this resource builds: the owning app's override, else default_level.

A framework-authored logger takes the owning app's framework_log_level when it is set. An app-authored logger always keeps default_level, since the app's own log_level already feeds App.config_log_level.

Source code in src/hassette/resources/base.py

def resolve_log_level(self, author: LogAuthor, default_level: LogLevel) -> LogLevel:
    """Level for a logger this resource builds: the owning app's override, else ``default_level``.

    A framework-authored logger takes the owning app's ``framework_log_level`` when it is set. An
    app-authored logger always keeps ``default_level``, since the app's own ``log_level`` already feeds
    `App.config_log_level`.
    """
    if author == "app":
        return default_level
    return self.owner_framework_log_level() or default_level

owner_framework_log_level() -> LogLevel | None

The owning app's framework_log_level, or None when no app owns this resource or it is unset.

Found the same way as owner_log_identity: App answers for itself, every other resource asks its parent.

Source code in src/hassette/resources/base.py

def owner_framework_log_level(self) -> LogLevel | None:
    """The owning app's ``framework_log_level``, or None when no app owns this resource or it is unset.

    Found the same way as `owner_log_identity`: App answers for itself, every other resource asks its
    parent.
    """
    if self.parent is None or self.parent is self:
        return None
    return self.parent.owner_framework_log_level()

owner_log_identity(author: LogAuthor) -> LogIdentity

Identity for a logger this resource builds: the given author plus the owning app instance.

The owner is the nearest App at or above this resource. App overrides this to name itself; every other resource asks its parent, so an App's children and their descendants carry the app, and framework resources reach the root and carry no owner.

Source code in src/hassette/resources/base.py

def owner_log_identity(self, author: LogAuthor) -> LogIdentity:
    """Identity for a logger this resource builds: the given ``author`` plus the owning app instance.

    The owner is the nearest App at or above this resource. App overrides this to name itself;
    every other resource asks its parent, so an App's children and their descendants carry the
    app, and framework resources reach the root and carry no owner.
    """
    # A root resource is its own parent (see _get_logger_name).
    if self.parent is None or self.parent is self:
        return LogIdentity(author=author)
    return self.parent.owner_log_identity(author)

add_child(child_class: type[_ResourceT], **kwargs: Any) -> _ResourceT

Create and add a child resource to this resource.

Parameters:

Name	Type	Description	Default
child_class	type[_ResourceT]	The class of the child resource to create.	required
**kwargs	Any	Keyword arguments to pass to the child resource's constructor.	{}

Returns:

Type	Description
_ResourceT	The created child resource.

Source code in src/hassette/resources/base.py

def add_child(self, child_class: type[_ResourceT], **kwargs: Any) -> _ResourceT:
    """Create and add a child resource to this resource.

    Args:
        child_class: The class of the child resource to create.
        **kwargs: Keyword arguments to pass to the child resource's constructor.

    Returns:
        The created child resource.
    """
    if "parent" in kwargs:
        raise ValueError("Cannot specify 'parent' argument when adding a child resource; it is set automatically.")
    child_class._validate_parent(self)

    inst = child_class(hassette=self.hassette, parent=self, **kwargs)
    self.children.append(inst)
    return inst

initialize() -> None async

Coordinator front door: joins or creates the resource's one owned initialization attempt.

This method itself never runs class-specific initialization work — see _initialize_body(), which every initialization attempt (direct calls and start()'s spawned joiner) ultimately runs exactly once per attempt. See coordinate_initialize() in hassette.resources.lifecycle for the full admission sequence (re-entry rejection, joining an active shutdown, restart refusal, task ownership).

Source code in src/hassette/resources/base.py

@final
async def initialize(self) -> None:
    """Coordinator front door: joins or creates the resource's one owned initialization attempt.

    This method itself never runs class-specific initialization work — see
    ``_initialize_body()``, which every initialization attempt (direct calls and ``start()``'s
    spawned joiner) ultimately runs exactly once per attempt. See
    ``coordinate_initialize()`` in ``hassette.resources.lifecycle`` for the full admission
    sequence (re-entry rejection, joining an active shutdown, restart refusal, task
    ownership).
    """
    await coordinate_initialize(self)

before_initialize() -> None async

Optional: prepare to accept new work, allocate sockets, queues, temp files, etc.

Source code in src/hassette/resources/base.py

async def before_initialize(self) -> None:
    """Optional: prepare to accept new work, allocate sockets, queues, temp files, etc."""
    pass

on_initialize() -> None async

Primary hook: perform your own initialization (sockets, queues, temp files…).

Source code in src/hassette/resources/base.py

async def on_initialize(self) -> None:
    """Primary hook: perform your own initialization (sockets, queues, temp files…)."""
    pass

after_initialize() -> None async

Optional: finalize initialization, signal readiness, etc.

Source code in src/hassette/resources/base.py

async def after_initialize(self) -> None:
    """Optional: finalize initialization, signal readiness, etc."""
    pass

shutdown() -> ShutdownOutcome async

Coordinator front door: joins or creates the resource's one owned shutdown attempt.

Returns the attempt's stored ShutdownOutcome.

This method itself never runs class-specific shutdown work — see _shutdown_body(), which the coordinator runs exactly once per attempt. Repeated calls after completion return the same stored outcome without rerunning hooks. See coordinate_shutdown() in hassette.resources.lifecycle for the full sequence (re-entry rejection, initializer cancellation/observation, bounded body observation, force-terminal outcome merging).

Source code in src/hassette/resources/base.py

@final
async def shutdown(self) -> ShutdownOutcome:
    """Coordinator front door: joins or creates the resource's one owned shutdown attempt.

    Returns the attempt's stored ``ShutdownOutcome``.

    This method itself never runs class-specific shutdown work — see ``_shutdown_body()``,
    which the coordinator runs exactly once per attempt. Repeated calls after completion
    return the same stored outcome without rerunning hooks. See ``coordinate_shutdown()`` in
    ``hassette.resources.lifecycle`` for the full sequence (re-entry rejection, initializer
    cancellation/observation, bounded body observation, force-terminal outcome merging).
    """
    return await coordinate_shutdown(self)

before_shutdown() -> None async

Optional: stop accepting new work, signal loops to wind down, etc.

Source code in src/hassette/resources/base.py

async def before_shutdown(self) -> None:
    """Optional: stop accepting new work, signal loops to wind down, etc."""
    pass

on_shutdown() -> None async

Primary hook: release your own stuff (sockets, queues, temp files…).

Source code in src/hassette/resources/base.py

async def on_shutdown(self) -> None:
    """Primary hook: release your own stuff (sockets, queues, temp files…)."""
    pass

after_shutdown() -> None async

Optional: last-chance actions after on_shutdown, before cleanup/STOPPED.

Source code in src/hassette/resources/base.py

async def after_shutdown(self) -> None:
    """Optional: last-chance actions after on_shutdown, before cleanup/STOPPED."""
    pass

cleanup() -> None async

Cleanup resources owned by the instance.

Called during shutdown, after the shutdown coordinator has already cancelled and observed any active initialization task, and after _run_task_bucket_shutdown_stage() has sealed and cancelled the TaskBucket's tracked work. Subclass overrides use this for resources they own directly (caches, connections, etc.); the base implementation owns nothing.

Source code in src/hassette/resources/base.py

async def cleanup(self) -> None:
    """Cleanup resources owned by the instance.

    Called during shutdown, after the shutdown coordinator has already cancelled and observed
    any active initialization task, and after ``_run_task_bucket_shutdown_stage()`` has sealed
    and cancelled the TaskBucket's tracked work. Subclass overrides use this for resources they
    own directly (caches, connections, etc.); the base implementation owns nothing.
    """
    self.lifecycle_logger.debug("Cleaned up resources")