"""Centralise mutable runtime application state. Runtime state is kept outside of Starlette's raw ``app.state`` storage and exposed through a controlled state manager object. This keeps the FastAPI framework state bag limited to shared infrastructure handles and immutable configuration while still allowing existing code to access runtime values via attribute proxying. """ from __future__ import annotations import datetime from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any from starlette.datastructures import State from app.models.server import ServerStatus if TYPE_CHECKING: # pragma: no cover from app.models.config import PendingRecovery ActivationRecord = dict[str, datetime.datetime] _RUNTIME_ATTRIBUTES: frozenset[str] = frozenset( { "setup_complete_cached", "server_status", "pending_recovery", "last_activation", } ) @dataclass class RuntimeState: """Mutable runtime state for the current application instance.""" setup_complete_cached: bool = False server_status: ServerStatus = field(default_factory=lambda: ServerStatus(online=False)) pending_recovery: PendingRecovery | None = None last_activation: ActivationRecord | None = None class ApplicationState(State): """Application state wrapper that delegates runtime state access. This allows runtime values to be stored in a dedicated :class:`RuntimeState` instance while preserving the familiar attribute-based ``app.state`` API for the rest of the application. """ def __init__(self, runtime_state: RuntimeState, state: dict[str, Any] | None = None): super().__init__(state) object.__setattr__(self, "_runtime_state", runtime_state) @property def runtime_state(self) -> RuntimeState: """Return the dedicated runtime state manager.""" return object.__getattribute__(self, "_runtime_state") def __getattr__(self, key: str) -> Any: if key in _RUNTIME_ATTRIBUTES: return getattr(self.runtime_state, key) return super().__getattr__(key) def __setattr__(self, key: str, value: Any) -> None: if key in _RUNTIME_ATTRIBUTES: setattr(self.runtime_state, key, value) return super().__setattr__(key, value) def __delattr__(self, key: str) -> None: if key in _RUNTIME_ATTRIBUTES: delattr(self.runtime_state, key) return super().__delattr__(key) def get_runtime_state(app: Any) -> RuntimeState: """Return the runtime state manager for the current FastAPI application.""" state = getattr(app, "state", None) if state is None or not hasattr(state, "runtime_state"): raise AttributeError("Runtime state has not been initialised on the application.") return state.runtime_state