88 lines
3.3 KiB
Python
88 lines
3.3 KiB
Python
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"))
|