solar_eval/cli/client.py

"""HTTP client wrapper for remote eval-engine API."""

import json as json_module
from typing import Any, Iterator

import click
import httpx


class EvalClient:
    """Synchronous HTTP client for the eval engine REST API."""

    def __init__(self, base_url: str, timeout: int = 300) -> None:
        self.base_url = base_url.rstrip("/")
        self._client = httpx.Client(timeout=timeout)

    def _url(self, path: str) -> str:
        return f"{self.base_url}{path}"

    def _handle_error(self, exc: httpx.HTTPStatusError) -> None:
        try:
            detail = exc.response.json().get("detail", exc.response.text)
        except Exception:
            detail = exc.response.text
        raise click.ClickException(f"API error ({exc.response.status_code}): {detail}")

    def get(self, path: str, params: dict[str, Any] | None = None) -> Any:
        resp = self._client.get(self._url(path), params=params)
        try:
            resp.raise_for_status()
        except httpx.HTTPStatusError as e:
            self._handle_error(e)
        return resp.json()

    def post(self, path: str, json: dict[str, Any] | None = None) -> Any:
        resp = self._client.post(self._url(path), json=json)
        try:
            resp.raise_for_status()
        except httpx.HTTPStatusError as e:
            self._handle_error(e)
        return resp.json()

    def put(self, path: str, json: dict[str, Any] | None = None) -> Any:
        resp = self._client.put(self._url(path), json=json)
        try:
            resp.raise_for_status()
        except httpx.HTTPStatusError as e:
            self._handle_error(e)
        return resp.json()

    def stream_sse(self, path: str, json: dict[str, Any] | None = None) -> Iterator[dict]:
        with self._client.stream("POST", self._url(path), json=json, timeout=None) as resp:
            for line in resp.iter_lines():
                if not line or not line.startswith("data: "):
                    continue
                data = line[len("data: "):]
                if data.strip():
                    yield json_module.loads(data)