# retoor <retoor@molodetz.nl>
from __future__ import annotations
import logging
from datetime import datetime, timezone
from typing import Annotated
from fastapi import APIRouter, Depends, Form, Request
from fastapi.responses import JSONResponse
from devplacepy.database import get_table
from devplacepy.responses import json_error, respond
from devplacepy.utils import generate_uid, get_current_user, is_admin, require_user
logger = logging.getLogger("routers.monitors")
router = APIRouter(tags=["monitors"])
@router.get("/checks")
async def list_checks(
request: Request,
user: dict = Depends(require_user),
):
checks_table = get_table("monitor_checks")
admin = is_admin(user)
if admin:
checks = list(checks_table.find(deleted_at=None, order_by="-created_at"))
else:
checks = list(
checks_table.find(user_uid=user["uid"], deleted_at=None, order_by="-created_at")
)
# Enrich with items count
items_table = get_table("monitor_check_items")
for check in checks:
try:
items = list(items_table.find(check_uid=check["uid"], deleted_at=None))
check["items_count"] = len(items)
except Exception:
check["items_count"] = 0
return respond(request, "monitors/list.html", {"checks": checks})
@router.post("/checks")
async def create_check(
request: Request,
user: dict = Depends(require_user),
name: Annotated[str, Form(max_length=128)] = ...,
check_type: Annotated[str, Form()] = ...,
target: Annotated[str, Form(max_length=1024)] = ...,
interval_seconds: Annotated[int, Form(ge=10, le=86400)] = 60,
description: Annotated[str, Form(max_length=512)] = "",
group_uid: Annotated[str, Form()] = "",
escalation_policy_uid: Annotated[str, Form()] = "",
):
# Validate target - reject private/internal addresses
if check_type == "http":
from urllib.parse import urlparse
parsed = urlparse(target)
host = parsed.hostname or ""
if host in ("localhost", "127.0.0.1", "0.0.0.0"):
return json_error(400, "Internal/private targets are not allowed")
if host.startswith("10.") or host.startswith("192.168.") or host.startswith("172.16."):
return json_error(400, "Private network targets are not allowed")
uid = generate_uid()
now = datetime.now(timezone.utc).isoformat()
checks_table = get_table("monitor_checks")
checks_table.insert(
{
"uid": uid,
"user_uid": user["uid"],
"name": name,
"description": description,
"check_type": check_type,
"target": target,
"interval_seconds": interval_seconds,
"group_uid": group_uid,
"escalation_policy_uid": escalation_policy_uid,
"enabled": 1,
"created_at": now,
"updated_at": now,
"deleted_at": None,
"deleted_by": None,
}
)
# Add default check item
from devplacepy.config import MOLOUPTIME_BINARY_AVAILABLE
if MOLOUPTIME_BINARY_AVAILABLE:
try:
item_uid = generate_uid()
items_table = get_table("monitor_check_items")
items_table.insert(
{
"uid": item_uid,
"check_uid": uid,
"user_uid": user["uid"],
"check_type": check_type,
"target": target,
"interval_seconds": interval_seconds,
"created_at": now,
"updated_at": now,
"deleted_at": None,
"deleted_by": None,
}
)
except Exception as e:
logger.error("Failed to create check item: %s", e)
return {"uid": uid, "name": name, "status": "created"}
@router.get("/checks/{check_uid}")
async def get_check(
request: Request,
check_uid: str,
user: dict = Depends(require_user),
):
checks_table = get_table("monitor_checks")
check = checks_table.find_one(uid=check_uid)
if not check:
return json_error(404, "Check not found")
if not is_admin(user) and check["user_uid"] != user["uid"]:
return json_error(403, "Access denied")
# Get items
items_table = get_table("monitor_check_items")
items = list(items_table.find(check_uid=check_uid, deleted_at=None))
check["items"] = items
return respond(request, "monitors/detail.html", {"check": check})
@router.delete("/checks/{check_uid}")
async def delete_check(
request: Request,
check_uid: str,
user: dict = Depends(require_user),
):
from devplacepy.database import soft_delete
checks_table = get_table("monitor_checks")
check = checks_table.find_one(uid=check_uid)
if not check:
return json_error(404, "Check not found")
if not is_admin(user) and check["user_uid"] != user["uid"]:
return json_error(403, "Access denied")
soft_delete("monitor_checks", check_uid, user["uid"])
# Also soft-delete items
items_table = get_table("monitor_check_items")
for item in items_table.find(check_uid=check_uid, deleted_at=None):
soft_delete("monitor_check_items", item["uid"], user["uid"])
return {"status": "deleted", "uid": check_uid}
@router.post("/checks/{check_uid}/toggle")
async def toggle_check(
request: Request,
check_uid: str,
user: dict = Depends(require_user),
):
checks_table = get_table("monitor_checks")
check = checks_table.find_one(uid=check_uid)
if not check:
return json_error(404, "Check not found")
if not is_admin(user) and check["user_uid"] != user["uid"]:
return json_error(403, "Access denied")
new_enabled = 0 if check.get("enabled", 1) else 1
checks_table.update(
{"uid": check_uid, "enabled": new_enabled, "updated_at": datetime.now(timezone.utc).isoformat()},
["uid"],
)
return {"status": "toggled", "enabled": new_enabled}
@router.get("/groups")
async def list_groups(
request: Request,
user: dict = Depends(require_user),
):
groups_table = get_table("monitor_groups")
admin = is_admin(user)
if admin:
groups = list(groups_table.find(deleted_at=None, order_by="-created_at"))
else:
groups = list(
groups_table.find(user_uid=user["uid"], deleted_at=None, order_by="-created_at")
)
return respond(request, "monitors/groups.html", {"groups": groups})
@router.post("/groups")
async def create_group(
request: Request,
user: dict = Depends(require_user),
name: Annotated[str, Form(max_length=128)] = ...,
description: Annotated[str, Form(max_length=512)] = "",
):
uid = generate_uid()
now = datetime.now(timezone.utc).isoformat()
groups_table = get_table("monitor_groups")
groups_table.insert(
{
"uid": uid,
"user_uid": user["uid"],
"name": name,
"description": description,
"created_at": now,
"deleted_at": None,
"deleted_by": None,
}
)
return {"uid": uid, "name": name, "status": "created"}
@router.get("/admin/stats")
async def admin_stats(
request: Request,
user: dict = Depends(require_user),
):
if not is_admin(user):
return json_error(403, "Admin access required")
checks_table = get_table("monitor_checks")
total_checks = 0
total_items = 0
by_type = {}
by_user = {}
try:
rows = checks_table.find(deleted_at=None)
for row in rows:
total_checks += 1
ct = row.get("check_type", "unknown")
by_type[ct] = by_type.get(ct, 0) + 1
uu = row.get("user_uid", "unknown")
by_user[uu] = by_user.get(uu, 0) + 1
except Exception:
pass
items_table = get_table("monitor_check_items")
try:
total_items = len(list(items_table.find(deleted_at=None)))
except Exception:
pass
snapshots_table = get_table("monitor_metrics_snapshots")
total_metrics = 0
try:
total_metrics = len(list(snapshots_table.find()))
except Exception:
pass
return respond(
request,
"admin/monitors.html",
{
"total_checks": total_checks,
"total_items": total_items,
"total_metrics": total_metrics,
"by_type": by_type,
"by_user": by_user,
},
)
@router.get("/escalation-policies")
async def list_escalation_policies(
request: Request,
user: dict = Depends(require_user),
):
policies_table = get_table("monitor_escalation_policies")
policies = list(
policies_table.find(user_uid=user["uid"], order_by="-created_at")
)
return respond(request, "monitors/escalation.html", {"policies": policies})
View Git Blame Copy Permalink