49 lines
1.2 KiB
Python
49 lines
1.2 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from typing import TYPE_CHECKING
|
||
|
|
|
||
|
|
from fastapi import FastAPI, HTTPException
|
||
|
|
|
||
|
|
if TYPE_CHECKING:
|
||
|
|
from .checks.base import Check
|
||
|
|
|
||
|
|
app = FastAPI(title="ha-diag-agent", version="0.1.0")
|
||
|
|
|
||
|
|
# Populated by main.py during startup
|
||
|
|
_checks: dict[str, "Check"] = {}
|
||
|
|
_node_name: str = "unknown"
|
||
|
|
_location_tag: str = "default"
|
||
|
|
|
||
|
|
|
||
|
|
def register_checks(checks: list["Check"], node_name: str, location_tag: str) -> None:
|
||
|
|
global _node_name, _location_tag
|
||
|
|
_checks.update({c.name: c for c in checks})
|
||
|
|
_node_name = node_name
|
||
|
|
_location_tag = location_tag
|
||
|
|
|
||
|
|
|
||
|
|
@app.get("/health")
|
||
|
|
async def health() -> dict:
|
||
|
|
return {
|
||
|
|
"status": "ok",
|
||
|
|
"node": _node_name,
|
||
|
|
"location_tag": _location_tag,
|
||
|
|
"checks": list(_checks.keys()),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
@app.post("/trigger/{check_name}")
|
||
|
|
async def trigger(check_name: str) -> dict:
|
||
|
|
check = _checks.get(check_name)
|
||
|
|
if check is None:
|
||
|
|
raise HTTPException(status_code=404, detail=f"Unknown check: {check_name!r}")
|
||
|
|
result = await check.run()
|
||
|
|
return {
|
||
|
|
"check": check_name,
|
||
|
|
"healthy": result.healthy,
|
||
|
|
"event_type": result.event_type,
|
||
|
|
"severity": result.severity,
|
||
|
|
"message": result.message,
|
||
|
|
"payload": result.payload,
|
||
|
|
}
|