from __future__ import annotations from quart import Blueprint, render_template, current_app from sqlalchemy import select, func from fablednetmon.auth.middleware import require_role from fablednetmon.models.users import UserRole from fablednetmon.models.hosts import Host from fablednetmon.models.monitors import PingResult, DnsResult dashboard_bp = Blueprint("dashboard", __name__) @dashboard_bp.get("/") @require_role(UserRole.viewer) async def index(): async with current_app.db_sessionmaker() as db: # Latest ping per host latest_ping_subq = ( select(PingResult.host_id, func.max(PingResult.probed_at).label("max_at")) .group_by(PingResult.host_id) .subquery() ) pr = await db.execute( select(PingResult).join( latest_ping_subq, (PingResult.host_id == latest_ping_subq.c.host_id) & (PingResult.probed_at == latest_ping_subq.c.max_at), ) ) latest_pings = list(pr.scalars()) ping_up = sum(1 for p in latest_pings if p.status.value == "up") ping_down = sum(1 for p in latest_pings if p.status.value == "down") # Latest DNS per host latest_dns_subq = ( select(DnsResult.host_id, func.max(DnsResult.resolved_at).label("max_at")) .group_by(DnsResult.host_id) .subquery() ) dr = await db.execute( select(DnsResult).join( latest_dns_subq, (DnsResult.host_id == latest_dns_subq.c.host_id) & (DnsResult.resolved_at == latest_dns_subq.c.max_at), ) ) latest_dns = list(dr.scalars()) dns_ok = sum(1 for d in latest_dns if d.status.value == "resolved") dns_fail = sum(1 for d in latest_dns if d.status.value != "resolved") total_hosts = (await db.execute(select(func.count()).select_from(Host))).scalar() poll_interval = current_app.config.get("MONITORS_POLL_INTERVAL", 60) return await render_template( "dashboard/index.html", ping_up=ping_up, ping_down=ping_down, dns_ok=dns_ok, dns_fail=dns_fail, total_hosts=total_hosts, poll_interval=poll_interval, )