FabledSteward/fabledscryer/hosts/routes.py

from __future__ import annotations
from quart import Blueprint, render_template, request, redirect, url_for, current_app
from sqlalchemy import select, func
from fabledscryer.auth.middleware import require_role
from fabledscryer.models.hosts import Host, ProbeType
from fabledscryer.models.monitors import PingResult, DnsResult
from fabledscryer.models.users import UserRole

hosts_bp = Blueprint("hosts", __name__, url_prefix="/hosts")


@hosts_bp.get("/")
@require_role(UserRole.viewer)
async def list_hosts():
    async with current_app.db_sessionmaker() as db:
        result = await db.execute(select(Host).order_by(Host.name))
        hosts = result.scalars().all()

        # Latest ping result 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 = {r.host_id: r for r in pr.scalars()}

        # Latest DNS result 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 = {r.host_id: r for r in dr.scalars()}

    return await render_template(
        "hosts/list.html",
        hosts=hosts,
        latest_pings=latest_pings,
        latest_dns=latest_dns,
    )


@hosts_bp.get("/new")
@require_role(UserRole.operator)
async def new_host():
    return await render_template("hosts/form.html", host=None, probe_types=list(ProbeType))


@hosts_bp.post("/")
@require_role(UserRole.operator)
async def create_host():
    form = await request.form
    host = Host(
        name=form["name"].strip(),
        address=form["address"].strip(),
        probe_type=ProbeType(form.get("probe_type", "tcp")),
        probe_port=int(form.get("probe_port") or 80),
        ping_enabled="ping_enabled" in form,
        dns_enabled="dns_enabled" in form,
        dns_expected_ip=form.get("dns_expected_ip", "").strip() or None,
        # poll_interval_seconds not exposed in UI yet; per-host scheduling not implemented
    )
    async with current_app.db_sessionmaker() as db:
        async with db.begin():
            db.add(host)
    return redirect(url_for("hosts.list_hosts"))


@hosts_bp.get("/<host_id>/edit")
@require_role(UserRole.operator)
async def edit_host(host_id: str):
    async with current_app.db_sessionmaker() as db:
        result = await db.execute(select(Host).where(Host.id == host_id))
        host = result.scalar_one_or_none()
    if host is None:
        return "Not found", 404
    return await render_template("hosts/form.html", host=host, probe_types=list(ProbeType))


@hosts_bp.post("/<host_id>")
@require_role(UserRole.operator)
async def update_host(host_id: str):
    form = await request.form
    async with current_app.db_sessionmaker() as db:
        async with db.begin():
            result = await db.execute(select(Host).where(Host.id == host_id))
            host = result.scalar_one_or_none()
            if host is None:
                return "Not found", 404
            host.name = form["name"].strip()
            host.address = form["address"].strip()
            host.probe_type = ProbeType(form.get("probe_type", "tcp"))
            host.probe_port = int(form.get("probe_port") or 80)
            host.ping_enabled = "ping_enabled" in form
            host.dns_enabled = "dns_enabled" in form
            host.dns_expected_ip = form.get("dns_expected_ip", "").strip() or None
            # poll_interval_seconds not exposed in UI yet
    return redirect(url_for("hosts.list_hosts"))


@hosts_bp.post("/<host_id>/delete")
@require_role(UserRole.admin)
async def delete_host(host_id: str):
    async with current_app.db_sessionmaker() as db:
        async with db.begin():
            result = await db.execute(select(Host).where(Host.id == host_id))
            host = result.scalar_one_or_none()
            if host:
                await db.delete(host)
    return redirect(url_for("hosts.list_hosts"))