Add graph explorer UI shell

This commit is contained in:
2026-05-18 17:11:36 +02:00
parent d2056c9046
commit d31b1376c8
7 changed files with 396 additions and 3 deletions

View File

@@ -3,6 +3,7 @@ from __future__ import annotations
import argparse
import json
import sys
from dataclasses import dataclass
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
@@ -10,6 +11,7 @@ from typing import Any
from urllib.parse import parse_qs, urlparse
from .graph_explorer import fabric_graph_explorer_manifest, fabric_graph_explorer_payload
from .graph_explorer_ui import graph_explorer_page
from .registry import (
RegistryError,
RegistryStore,
@@ -24,6 +26,11 @@ from .registry import (
)
@dataclass(frozen=True)
class HtmlResponse:
body: str
class RegistryHandler(BaseHTTPRequestHandler):
store: RegistryStore
@@ -40,6 +47,8 @@ class RegistryHandler(BaseHTTPRequestHandler):
parts = _parts(path)
if path == "/health":
return HTTPStatus.OK, {"status": "ok"}
if parts == ["ui", "graph-explorer"]:
return HTTPStatus.OK, HtmlResponse(graph_explorer_page())
if path == "/status":
return HTTPStatus.OK, self.store.status()
if parts == ["repositories"]:
@@ -129,7 +138,10 @@ class RegistryHandler(BaseHTTPRequestHandler):
query = parse_qs(parsed.query)
try:
status, body = action(parsed.path, query)
self._send_json(int(status), body)
if isinstance(body, HtmlResponse):
self._send_text(int(status), body.body, "text/html; charset=utf-8")
else:
self._send_json(int(status), body)
except RegistryError as exc:
self._send_json(exc.status_code, {"error": exc.message})
except json.JSONDecodeError as exc:
@@ -149,8 +161,14 @@ class RegistryHandler(BaseHTTPRequestHandler):
def _send_json(self, status: int, body: Any) -> None:
payload = json.dumps(body, indent=2, sort_keys=True).encode("utf-8")
self._send_bytes(status, payload, "application/json; charset=utf-8")
def _send_text(self, status: int, body: str, content_type: str) -> None:
self._send_bytes(status, body.encode("utf-8"), content_type)
def _send_bytes(self, status: int, payload: bytes, content_type: str) -> None:
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)