from typing import Any from fastapi import APIRouter, Form from fastapi.responses import HTMLResponse from app import journal from app.render import render from app.systemd import units router = APIRouter(prefix="/api/services", tags=["services"]) SORT_KEYS = ("name", "state", "enabled") _STATE_RANK = { "active": 0, "activating": 1, "deactivating": 1, "reloading": 1, "reactivating": 1, "failed": 2, "inactive": 3, "dead": 3, "maintenance": 3, } _ENABLED_RANK = { "enabled": 0, "indirect": 1, "static": 2, "disabled": 3, "alias": 4, "linked": 4, "linked-runtime": 4, "masked": 5, "": 6, } def _rank(u: dict[str, Any], key: str) -> int: if key == "state": return _STATE_RANK.get(u["active"], 9) if key == "enabled": return _ENABLED_RANK.get(u["enabled"], 9) return 0 async def _list_fragment(q: str, sort: str = "name", order: str = "asc", error: str | None = None) -> str: if sort not in SORT_KEYS: sort = "name" if order not in ("asc", "desc"): order = "asc" unit_list = await units.unit_list() if q: ql = q.lower() unit_list = [ u for u in unit_list if ql in u["name"].lower() or ql in u["desc"].lower() ] reverse = order == "desc" if sort == "name": unit_list.sort(key=lambda u: u["name"], reverse=reverse) else: unit_list.sort(key=lambda u: (_rank(u, sort), u["name"]), reverse=reverse) state = await units.system_state() return render( "services.html", units=unit_list, state=state, q=q, sort=sort, order=order, error=error, ) @router.get("") async def services(q: str = "", sort: str = "name", order: str = "asc"): return HTMLResponse(await _list_fragment(q, sort, order)) @router.get("/{unit}/detail") async def service_detail(unit: str): error = None props: dict[str, str] = {} log: list[dict[str, str]] = [] try: props = await units.unit_detail(unit) except (ValueError, RuntimeError) as e: error = str(e)[:300] if not error: try: log, _ = await journal.tail(None, "all", unit, None, 15) except (RuntimeError, OSError): pass return HTMLResponse(render("service_detail.html", unit=unit, props=props, log=log, error=error)) @router.post("/{unit}/action") async def service_action( unit: str, action: str = Form(...), q: str = Form(""), sort: str = Form("name"), order: str = Form("asc"), ): error = None try: _ = await units.unit_action(unit, action) except ValueError as e: error = str(e) except RuntimeError as e: error = str(e)[:300] return HTMLResponse(await _list_fragment(q, sort, order, error=error))