generated from coulomb/repo-seed
112 lines
3.8 KiB
Python
112 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from http import HTTPStatus
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
from .service import (
|
|
CanonServiceError,
|
|
artifact_graph,
|
|
inspect_canon,
|
|
list_artifacts,
|
|
list_models,
|
|
list_standards,
|
|
list_views,
|
|
profile_inspect,
|
|
read_view,
|
|
validate_canon,
|
|
)
|
|
|
|
|
|
def serve(host: str = "127.0.0.1", port: int = 8765, root: Path | None = None) -> None:
|
|
handler = _build_handler(root)
|
|
server = ThreadingHTTPServer((host, port), handler)
|
|
print(f"InfoTechCanon API listening on http://{host}:{port}", flush=True)
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
server.server_close()
|
|
|
|
|
|
def _build_handler(root: Path | None) -> type[BaseHTTPRequestHandler]:
|
|
class CanonRequestHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None:
|
|
parsed = urlparse(self.path)
|
|
query = parse_qs(parsed.query)
|
|
try:
|
|
status, payload = _route(parsed.path, query, root)
|
|
except CanonServiceError as exc:
|
|
status, payload = HTTPStatus.BAD_REQUEST, exc.to_dict()
|
|
except Exception as exc:
|
|
status, payload = HTTPStatus.INTERNAL_SERVER_ERROR, {
|
|
"ok": False,
|
|
"error": {
|
|
"code": "unhandled_error",
|
|
"message": str(exc),
|
|
"details": {},
|
|
},
|
|
}
|
|
self._send_json(status, payload)
|
|
|
|
def log_message(self, format: str, *args: object) -> None:
|
|
return
|
|
|
|
def _send_json(self, status: HTTPStatus, payload: dict[str, Any]) -> None:
|
|
body = json.dumps(payload, indent=2, sort_keys=True).encode("utf-8")
|
|
self.send_response(status.value)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
return CanonRequestHandler
|
|
|
|
|
|
def _route(
|
|
path: str,
|
|
query: dict[str, list[str]],
|
|
root: Path | None,
|
|
) -> tuple[HTTPStatus, dict[str, Any]]:
|
|
if path == "/health":
|
|
return HTTPStatus.OK, {"ok": True, "service": "info-tech-canon"}
|
|
if path == "/inspect":
|
|
return HTTPStatus.OK, inspect_canon(root)
|
|
if path == "/artifacts":
|
|
return HTTPStatus.OK, list_artifacts(root, kind=_first(query, "kind"))
|
|
if path == "/models":
|
|
return HTTPStatus.OK, list_models(root)
|
|
if path == "/standards":
|
|
return HTTPStatus.OK, list_standards(root)
|
|
if path == "/validate":
|
|
payload = validate_canon(root)
|
|
return (HTTPStatus.OK if payload["ok"] else HTTPStatus.BAD_REQUEST), payload
|
|
if path == "/graph":
|
|
graph_format = _first(query, "format") or "json"
|
|
return HTTPStatus.OK, artifact_graph(root, output_format=graph_format)
|
|
if path == "/views":
|
|
return HTTPStatus.OK, list_views(root)
|
|
if path.startswith("/views/"):
|
|
name = path.removeprefix("/views/").strip("/")
|
|
return HTTPStatus.OK, read_view(name, root)
|
|
if path.startswith("/profiles/") and path.endswith("/inspect"):
|
|
profile = path.removeprefix("/profiles/").removesuffix("/inspect").strip("/")
|
|
return HTTPStatus.OK, profile_inspect(profile, root)
|
|
return HTTPStatus.NOT_FOUND, {
|
|
"ok": False,
|
|
"error": {
|
|
"code": "not_found",
|
|
"message": f"Unknown endpoint: {path}",
|
|
"details": {"path": path},
|
|
},
|
|
}
|
|
|
|
|
|
def _first(query: dict[str, list[str]], name: str) -> str | None:
|
|
values = query.get(name) or []
|
|
return values[0] if values else None
|