110 lines
2.8 KiB
Python
110 lines
2.8 KiB
Python
from typing import Annotated, 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: Annotated[str, Form()],
|
|
q: Annotated[str, Form()] = "",
|
|
sort: Annotated[str, Form()] = "name",
|
|
order: Annotated[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))
|