117 lines
3.0 KiB
Python
117 lines
3.0 KiB
Python
"""Basic health check for the SmartTools Registry API and DB."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import sqlite3
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import URLError, HTTPError
|
|
|
|
|
|
def _add_src_to_path() -> None:
|
|
repo_root = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(repo_root / "src"))
|
|
|
|
|
|
def _get_registry_url() -> str:
|
|
from smarttools.config import get_registry_url
|
|
|
|
return get_registry_url().rstrip("/")
|
|
|
|
|
|
def _get_db_path() -> Path:
|
|
from smarttools.registry.db import get_db_path
|
|
|
|
return get_db_path()
|
|
|
|
|
|
def check_api(url: str) -> str | None:
|
|
endpoint = f"{url}/tools?limit=1"
|
|
try:
|
|
req = Request(endpoint, headers={"Accept": "application/json"})
|
|
with urlopen(req, timeout=5) as resp:
|
|
if resp.status != 200:
|
|
return f"API unhealthy: status {resp.status}"
|
|
payload = json.loads(resp.read().decode("utf-8"))
|
|
if "data" not in payload:
|
|
return "API unhealthy: missing data field"
|
|
except HTTPError as exc:
|
|
return f"API unhealthy: HTTP {exc.code}"
|
|
except URLError as exc:
|
|
return f"API unhealthy: {exc.reason}"
|
|
except Exception as exc:
|
|
return f"API unhealthy: {exc}"
|
|
return None
|
|
|
|
|
|
def check_db(db_path: Path) -> str | None:
|
|
if not db_path.exists():
|
|
return f"DB missing: {db_path}"
|
|
if not db_path.is_file():
|
|
return f"DB path is not a file: {db_path}"
|
|
try:
|
|
with db_path.open("rb"):
|
|
pass
|
|
except OSError as exc:
|
|
return f"DB not readable: {exc}"
|
|
return None
|
|
|
|
|
|
def check_last_sync(db_path: Path) -> str | None:
|
|
try:
|
|
conn = sqlite3.connect(db_path)
|
|
try:
|
|
row = conn.execute("SELECT MAX(processed_at) FROM webhook_log").fetchone()
|
|
finally:
|
|
conn.close()
|
|
except sqlite3.Error as exc:
|
|
return f"DB error: {exc}"
|
|
|
|
if not row or not row[0]:
|
|
return "No webhook sync recorded in webhook_log"
|
|
|
|
raw = str(row[0]).replace("Z", "+00:00")
|
|
try:
|
|
last_sync = datetime.fromisoformat(raw)
|
|
except ValueError:
|
|
return f"Invalid sync timestamp: {row[0]}"
|
|
|
|
if datetime.utcnow() - last_sync > timedelta(hours=1):
|
|
return f"Last sync too old: {last_sync.isoformat()}"
|
|
return None
|
|
|
|
|
|
def main() -> int:
|
|
_add_src_to_path()
|
|
errors = []
|
|
|
|
registry_url = _get_registry_url()
|
|
db_path = _get_db_path()
|
|
|
|
api_error = check_api(registry_url)
|
|
if api_error:
|
|
errors.append(api_error)
|
|
|
|
db_error = check_db(db_path)
|
|
if db_error:
|
|
errors.append(db_error)
|
|
else:
|
|
sync_error = check_last_sync(db_path)
|
|
if sync_error:
|
|
errors.append(sync_error)
|
|
|
|
if errors:
|
|
for err in errors:
|
|
print(err, file=sys.stderr)
|
|
return 1
|
|
|
|
print("Health check OK")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|