feat: host management UI (CRUD)

fablednetmon/hosts/routes.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+from quart import Blueprint, render_template, request, redirect, url_for, current_app
+from sqlalchemy import select
+from fablednetmon.auth.middleware import require_role
+from fablednetmon.models.hosts import Host, ProbeType
+from fablednetmon.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()
+    return await render_template("hosts/list.html", hosts=hosts)
+
+
+@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"))