File size: 2,146 Bytes
4a6ae6c | 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 WarehouseEye FastAPI backend."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import httpx
class BackendUnavailableError(RuntimeError):
"""Raised when API backend cannot be reached."""
@dataclass
class ApiClient:
"""Small sync API client tailored for Streamlit interactions."""
base_url: str
timeout_sec: float = 4.0
def _request(self, method: str, path: str, json_data: dict[str, Any] | None = None) -> dict[str, Any]:
url = f"{self.base_url.rstrip('/')}{path}"
try:
response = httpx.request(
method=method,
url=url,
json=json_data,
timeout=self.timeout_sec,
)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as exc:
detail = exc.response.text
raise RuntimeError(f"{exc.response.status_code} {exc.response.reason_phrase}: {detail}") from exc
except httpx.HTTPError as exc:
raise BackendUnavailableError(
"WarehouseEye backend is unavailable. Start FastAPI and verify API_BASE_URL."
) from exc
def get_health(self) -> dict[str, Any]:
return self._request("GET", "/health")
def get_timeline(self, video_id: str) -> dict[str, Any]:
return self._request("GET", f"/timeline/{video_id}")
def get_track_timeline(self, video_id: str, track_id: int) -> dict[str, Any]:
return self._request("GET", f"/timeline/{video_id}/track/{track_id}")
def analyze(self, video_id: str, video_url: str, force: bool = False) -> dict[str, Any]:
# API currently requires video_url in AnalyzeRequest schema.
return self._request("POST", "/analyze", {"video_id": video_id, "video_url": video_url, "force": force})
def query(self, video_id: str, question: str) -> dict[str, Any]:
return self._request("POST", "/query", {"video_id": video_id, "question": question})
def benchmark(self) -> dict[str, Any]:
return self._request("GET", "/benchmark")
|