137 lines
4.5 KiB
Python
137 lines
4.5 KiB
Python
"""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.config import PendingRecovery
|
|
from app.models.server import ServerStatus
|
|
|
|
if TYPE_CHECKING: # pragma: no cover
|
|
from app.config import Settings
|
|
|
|
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
|
|
|
|
|
|
def get_app_settings(app: Any) -> Settings:
|
|
"""Return the current immutable settings from application state."""
|
|
settings = getattr(app.state, "settings", None)
|
|
if settings is None:
|
|
raise AttributeError("Application settings are not available on the app state.")
|
|
return settings
|
|
|
|
|
|
def update_app_settings(app: Any, **overrides: Any) -> None:
|
|
"""Update the current application settings immutably."""
|
|
settings = get_app_settings(app)
|
|
app.state.settings = settings.model_copy(update=overrides)
|
|
|
|
|
|
def record_activation(app: Any, jail_name: str, at: datetime.datetime | None = None) -> datetime.datetime:
|
|
"""Record a jail activation timestamp in runtime state."""
|
|
now = at if at is not None else datetime.datetime.now(tz=datetime.UTC)
|
|
runtime_state = get_runtime_state(app)
|
|
runtime_state.last_activation = {
|
|
"jail_name": jail_name,
|
|
"at": now,
|
|
}
|
|
return now
|
|
|
|
|
|
def create_pending_recovery(
|
|
app: Any,
|
|
jail_name: str,
|
|
activated_at: datetime.datetime,
|
|
detected_at: datetime.datetime | None = None,
|
|
) -> None:
|
|
"""Create a pending recovery record in runtime state."""
|
|
runtime_state = get_runtime_state(app)
|
|
runtime_state.pending_recovery = PendingRecovery(
|
|
jail_name=jail_name,
|
|
activated_at=activated_at,
|
|
detected_at=detected_at if detected_at is not None else datetime.datetime.now(tz=datetime.UTC),
|
|
)
|
|
|
|
|
|
def clear_pending_recovery(app: Any) -> None:
|
|
"""Clear the current pending recovery record."""
|
|
get_runtime_state(app).pending_recovery = None
|
|
|
|
|
|
def clear_activation_record(app: Any) -> None:
|
|
"""Clear the current activation tracking record."""
|
|
get_runtime_state(app).last_activation = None
|