Spaces:
Running
Running
File size: 2,091 Bytes
9c84f9d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 | """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)
|