info-tech-canon/src/info_tech_canon/api.py

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,
    alignment_template,
    artifact_graph,
    inspect_canon,
    list_artifacts,
    list_models,
    list_standards,
    list_views,
    profile_graph,
    profile_inspect,
    profile_validate,
    read_view,
    review_capability_record,
    review_kit,
    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 == "/review-kit":
        return HTTPStatus.OK, review_kit(root)
    if path == "/alignment-template":
        return HTTPStatus.OK, alignment_template(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)
    if path.startswith("/profiles/") and path.endswith("/validate"):
        profile = path.removeprefix("/profiles/").removesuffix("/validate").strip("/")
        payload = profile_validate(profile, root)
        return (HTTPStatus.OK if payload["ok"] else HTTPStatus.BAD_REQUEST), payload
    if path == "/capability-review":
        record = _first(query, "path")
        if not record:
            return HTTPStatus.BAD_REQUEST, {
                "ok": False,
                "error": {
                    "code": "missing_path",
                    "message": "Query parameter path= is required",
                    "details": {},
                },
            }
        return HTTPStatus.OK, review_capability_record(record, root)
    if path.startswith("/profiles/") and path.endswith("/graph"):
        profile = path.removeprefix("/profiles/").removesuffix("/graph").strip("/")
        graph_format = _first(query, "format") or "json"
        return HTTPStatus.OK, profile_graph(profile, root, output_format=graph_format)
    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