feat: add openapi generator and python client stub

Author: haasonsaas

README.md

@@ -156,6 +156,11 @@ Example ingestion payload:
 - Set `PROVENANCE_DASHBOARD_API` to point at your deployed API when running remotely.
 - To enable trend charts, set `PROVENANCE_DASHBOARD_EVENTS` to a path containing the exported JSONL events (defaults to `data/timeseries_events.jsonl`).
 
+## SDK & Schema
+
+- Generate an OpenAPI schema with `make docs` (writes `openapi.json`).
+- A lightweight synchronous client lives in `clients/python`; use `ProvenanceClient` for basic ingestion/status/analytics calls.
+
 ## Data Persistence Model
 
 - **Analyses** – Stored as JSON blobs keyed by `analysis:{analysis_id}` with a sorted set index for time-window queries.

clients/python/__init__.py

@@ -0,0 +1,5 @@
+"""Python client stubs for interacting with the Provenance API."""
+
+from .client import ProvenanceClient, AnalysisRequest
+
+__all__ = ["ProvenanceClient", "AnalysisRequest"]

clients/python/client.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field
+from typing import Any, Dict, Iterable
+
+import httpx
+
+
+@dataclass
+class AnalysisRequest:
+    """Convenience wrapper for POST /analysis payloads."""
+
+    repo: str
+    pr_number: str
+    base_sha: str
+    head_sha: str
+    branch: str | None = None
+    provenance_data: Dict[str, Any] = field(default_factory=lambda: {"changed_lines": []})
+
+
+class ProvenanceClient:
+    """Lightweight synchronous client for the Provenance API."""
+
+    def __init__(
+        self,
+        base_url: str,
+        *,
+        timeout: float = 10.0,
+        headers: Dict[str, str] | None = None,
+        transport: httpx.BaseTransport | None = None,
+    ) -> None:
+        normalized_base = base_url.rstrip("/") + "/"
+        self._client = httpx.Client(
+            base_url=normalized_base,
+            timeout=timeout,
+            headers=headers,
+            transport=transport,
+        )
+
+    def __enter__(self) -> "ProvenanceClient":
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> None:
+        self.close()
+
+    def close(self) -> None:
+        self._client.close()
+
+    def submit_analysis(self, request: AnalysisRequest) -> dict:
+        response = self._client.post("analysis", json=asdict(request))
+        response.raise_for_status()
+        return response.json()
+
+    def get_analysis_status(self, analysis_id: str) -> dict:
+        response = self._client.get(f"analysis/{analysis_id}")
+        response.raise_for_status()
+        return response.json()
+
+    def get_analysis_decision(self, analysis_id: str) -> dict:
+        response = self._client.get(f"analysis/{analysis_id}/decision")
+        response.raise_for_status()
+        return response.json()
+
+    def get_analytics_summary(
+        self,
+        metric: str,
+        *,
+        time_window: str = "7d",
+        group_by: str = "agent_id",
+        category: str | None = None,
+        agent_id: str | None = None,
+    ) -> dict:
+        params = {
+            "metric": metric,
+            "time_window": time_window,
+            "group_by": group_by,
+        }
+        if category:
+            params["category"] = category
+        if agent_id:
+            params["agent_id"] = agent_id
+        response = self._client.get("analytics/summary", params=params)
+        response.raise_for_status()
+        return response.json()
+
+    def get_agent_behavior(self, *, time_window: str = "7d", agent_id: str | None = None, top_categories: int = 3) -> dict:
+        params = {"time_window": time_window, "top_categories": top_categories}
+        if agent_id:
+            params["agent_id"] = agent_id
+        response = self._client.get("analytics/agents/behavior", params=params)
+        response.raise_for_status()
+        return response.json()
+
+    def healthcheck(self) -> dict:
+        response = self._client.get("healthz")
+        response.raise_for_status()
+        return response.json()

scripts/generate_openapi.py

@@ -0,0 +1,36 @@
+"""Generate and persist the OpenAPI schema for the Provenance service."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from fastapi.openapi.utils import get_openapi
+
+from app.main import create_app
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Generate OpenAPI schema")
+    parser.add_argument(
+        "--output",
+        default="openapi.json",
+        help="Path to write the OpenAPI schema (default: openapi.json)",
+    )
+    args = parser.parse_args()
+
+    app = create_app()
+    schema = get_openapi(
+        title=app.title,
+        version=app.version,
+        routes=app.routes,
+        description=app.description,
+    )
+    output_path = Path(args.output)
+    output_path.write_text(json.dumps(schema, indent=2), encoding="utf-8")
+    print(f"OpenAPI schema written to {output_path}")
+
+
+if __name__ == "__main__":
+    main()

tests/test_client_stub.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import json
+
+import httpx
+
+from clients.python import AnalysisRequest, ProvenanceClient
+
+
+def test_client_submit_analysis_payload():
+    captured = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["method"] = request.method
+        captured["url"] = str(request.url)
+        captured["json"] = json.loads(request.content.decode())
+        return httpx.Response(202, json={"analysis_id": "an_1", "status": "received"})
+
+    transport = httpx.MockTransport(handler)
+    with ProvenanceClient("http://example.com/v1", transport=transport) as client:
+        request = AnalysisRequest(
+            repo="acme/repo",
+            pr_number="42",
+            base_sha="base",
+            head_sha="head",
+            provenance_data={"changed_lines": []},
+        )
+        response = client.submit_analysis(request)
+
+    assert response["analysis_id"] == "an_1"
+    assert captured["method"] == "POST"
+    assert captured["url"].endswith("/analysis")
+    assert captured["json"]["repo"] == "acme/repo"
+
+
+def test_client_summary_queries():
+    params_captured = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        params_captured.update(request.url.params)
+        return httpx.Response(200, json={"result": {"data": []}})
+
+    transport = httpx.MockTransport(handler)
+    with ProvenanceClient("http://example.com/v1", transport=transport) as client:
+        client.get_analytics_summary("risk_rate", time_window="3d", category="sqli", agent_id="claude")
+
+    assert params_captured["metric"] == "risk_rate"
+    assert params_captured["time_window"] == "3d"
+    assert params_captured["category"] == "sqli"
+    assert params_captured["agent_id"] == "claude"

tests/test_openapi_generation.py

@@ -0,0 +1,14 @@
+import json
+import sys
+from pathlib import Path
+
+import scripts.generate_openapi as generator
+
+
+def test_generate_openapi(tmp_path, monkeypatch):
+    output = tmp_path / "schema.json"
+    monkeypatch.setattr(sys, "argv", ["generate_openapi", "--output", str(output)])
+    generator.main()
+    assert output.exists()
+    schema = json.loads(output.read_text(encoding="utf-8"))
+    assert schema["info"]["title"] == "Provenance & Risk Analytics"