from __future__ import annotations from quart import Blueprint, current_app, render_template, request from sqlalchemy import func, select from steward.alerts.service import active_alerts, alert_label from steward.auth.middleware import require_role from steward.core.status import collect_status, sparkline_svg from steward.models.hosts import Host from steward.models.users import UserRole status_bp = Blueprint("status", __name__, url_prefix="/status") def _summarize(entries) -> tuple[int, int, int]: up = sum(1 for e in entries if e.status == "up") down = sum(1 for e in entries if e.status == "down") pending = sum(1 for e in entries if e.status == "pending") return up, down, pending @status_bp.get("/") @require_role(UserRole.viewer) async def index(): poll_interval = current_app.config.get("MONITORS_POLL_INTERVAL", 60) return await render_template("status/index.html", poll_interval=poll_interval) @status_bp.get("/rows") @require_role(UserRole.viewer) async def rows(): """HTMX fragment — what is wrong (down monitors, active alerts), then the full status list, refreshed on poll.""" async with current_app.db_sessionmaker() as db: entries = await collect_status(db) alerts = await active_alerts(db) for e in entries: if e.spark: e.spark_svg = sparkline_svg(e.spark) up, down, pending = _summarize(entries) return await render_template( "status/rows.html", entries=entries, up=up, down=down, pending=pending, alerts=alerts, problems=[e for e in entries if e.status == "down"], ) @status_bp.get("/widget") @require_role(UserRole.viewer) async def widget(): """HTMX dashboard 'Overview' widget — the single at-a-glance summary that replaced the old fixed top strip: host count + monitor up/down/pending + problem list + an alerts link.""" async with current_app.db_sessionmaker() as db: entries = await collect_status(db) total_hosts = (await db.execute( select(func.count()).select_from(Host))).scalar() alerts = await active_alerts(db) up, down, pending = _summarize(entries) return await render_template( "status/widget.html", entries=entries, up=up, down=down, pending=pending, total_hosts=total_hosts, alerts=overview_alerts(alerts), widget_id=request.args.get("wid", "0"), ) def overview_alerts(alerts) -> list[dict]: """The Overview's alert lines: breaching or silent first, then those someone has taken; each {rule, label, since}.""" order = {"firing": 0, "no_data": 1, "pending": 2, "acknowledged": 3} rows = [] for state, rule in alerts: label = alert_label(state) since = state.no_data_since if label == "no_data" else state.last_transition_at rows.append({"rule": rule, "label": label, "since": since}) return sorted(rows, key=lambda r: order.get(r["label"], 9))