from __future__ import annotations
"""Reports resource for the CorePlexML SDK."""
from coreplexml._http import HTTPClient
[docs]
class ReportsResource:
"""Generate and manage reports.
Reports provide downloadable PDF/HTML summaries of experiments,
models, deployments, project status, and SynthGen analysis.
"""
def __init__(self, http: HTTPClient):
self._http = http
@staticmethod
def _normalize_report(payload: dict) -> dict:
"""Normalize API payloads to a direct report object."""
if not isinstance(payload, dict):
return {}
report = payload.get("report")
if isinstance(report, dict):
return report
out = dict(payload)
if "id" not in out and out.get("report_id"):
out["id"] = out["report_id"]
return out
[docs]
def list(self, project_id: str | None = None, kind: str | None = None, limit: int = 50, offset: int = 0) -> dict:
"""List reports, optionally filtered by project or kind.
Args:
project_id: Filter by project UUID (optional).
kind: Filter by report type (optional).
limit: Maximum results (default 50).
offset: Pagination offset.
Returns:
Dictionary with ``items`` list and ``total`` count.
"""
params: dict = {"limit": limit, "offset": offset}
if project_id:
params["project_id"] = project_id
if kind:
params["kind"] = kind
return self._http.get("/api/reports", params=params)
[docs]
def create(self, project_id: str, kind: str, entity_id: str, options: dict | None = None) -> dict:
"""Create a new report.
Args:
project_id: UUID of the project.
kind: Report type -- ``project``, ``experiment``, ``model``, ``deployment``, or ``synthgen``.
entity_id: UUID of the entity to report on.
options: Optional report configuration (for example ``{"llm_insights": True}``).
Returns:
Created report dictionary with ``id`` and ``status``.
"""
body = {"project_id": project_id, "kind": kind, "entity_id": entity_id}
if options:
body["options"] = options
data = self._http.post("/api/reports", json=body)
return self._normalize_report(data)
[docs]
def get(self, report_id: str) -> dict:
"""Get report details.
Args:
report_id: UUID of the report.
Returns:
Report dictionary.
"""
data = self._http.get(f"/api/reports/{report_id}")
return self._normalize_report(data)
[docs]
def wait(self, report_id: str, interval: float = 3.0, timeout: float = 300.0) -> dict:
"""Poll report until generation completes.
Args:
report_id: UUID of the report.
interval: Seconds between polls (default 3.0).
timeout: Maximum seconds to wait (default 300.0).
Returns:
Final report status dictionary.
Raises:
CorePlexMLError: If the report times out.
"""
import time
start = time.time()
while time.time() - start < timeout:
raw = self._http.get(f"/api/reports/{report_id}")
report = self._normalize_report(raw)
status = report.get("status", "")
if status in ("succeeded", "completed", "failed", "error"):
return {"report": report}
time.sleep(interval)
from coreplexml.exceptions import CorePlexMLError
raise CorePlexMLError(f"Report {report_id} timed out after {timeout}s")
[docs]
def download(self, report_id: str, output_path: str) -> str:
"""Download a generated report to a local file.
Args:
report_id: UUID of the report.
output_path: Local path to save the file.
Returns:
The output_path on success.
"""
return self._http.download(
f"/api/reports/{report_id}/download",
output_path,
)