| """ |
| Thin HTTP client for the private backend. |
| |
| All TM logic lives behind the backend; this client just forwards requests and |
| surfaces errors. Configuration comes from environment variables: |
| |
| BACKEND_URL - base URL of the private backend (e.g. https://user-tm-backend.hf.space) |
| API_TOKEN - shared secret sent as the `X-API-Token` header (optional) |
| HF_TOKEN - HuggingFace token sent as `Authorization: Bearer` so a *private* |
| backend Space can be reached (optional) |
| """ |
|
|
| import os |
| from typing import Any, Dict, Optional |
|
|
| import requests |
|
|
| BACKEND_URL = os.environ.get("BACKEND_URL", "http://localhost:8000").rstrip("/") |
| API_TOKEN = os.environ.get("API_TOKEN", "").strip() |
| HF_TOKEN = os.environ.get("HF_TOKEN", "").strip() |
|
|
| _TIMEOUT = 600 |
|
|
|
|
| class BackendError(Exception): |
| pass |
|
|
|
|
| def _headers() -> Dict[str, str]: |
| h: Dict[str, str] = {} |
| if API_TOKEN: |
| h["X-API-Token"] = API_TOKEN |
| if HF_TOKEN: |
| h["Authorization"] = f"Bearer {HF_TOKEN}" |
| return h |
|
|
|
|
| def _request(method: str, path: str, **kwargs) -> Any: |
| url = f"{BACKEND_URL}{path}" |
| try: |
| resp = requests.request(method, url, headers=_headers(), timeout=_TIMEOUT, **kwargs) |
| except requests.RequestException as e: |
| raise BackendError(f"Could not reach backend at {BACKEND_URL}: {e}") |
| if resp.status_code >= 400: |
| try: |
| detail = resp.json().get("detail", resp.text) |
| except Exception: |
| detail = resp.text |
| raise BackendError(detail) |
| return resp.json() |
|
|
|
|
| def health() -> bool: |
| try: |
| return _request("GET", "/health").get("status") == "ok" |
| except BackendError: |
| return False |
|
|
|
|
| def get_datasets() -> Dict[str, Any]: |
| return _request("GET", "/datasets") |
|
|
|
|
| def train(tm_type: str, dataset_key: str, n_trials: int, epochs: int) -> Dict[str, Any]: |
| return _request("POST", "/train", json={ |
| "tm_type": tm_type, "dataset_key": dataset_key, |
| "n_trials": n_trials, "epochs": epochs, |
| }) |
|
|
|
|
| def upload(model_bytes: bytes, model_name: str, csv_bytes: bytes, csv_name: str) -> Dict[str, Any]: |
| files = { |
| "model_json": (model_name, model_bytes, "application/json"), |
| "samples_csv": (csv_name, csv_bytes, "text/csv"), |
| } |
| return _request("POST", "/upload", files=files) |
|
|
|
|
| def meta(model_id: str) -> Dict[str, Any]: |
| return _request("POST", "/meta", json={"model_id": model_id}) |
|
|
|
|
| def visualize(payload: Dict[str, Any]) -> Dict[str, Any]: |
| return _request("POST", "/visualize", json=payload) |
|
|
|
|
| def scatter(payload: Dict[str, Any]) -> Dict[str, Any]: |
| return _request("POST", "/scatter", json=payload) |
|
|
|
|
| def export(payload: Dict[str, Any]) -> Dict[str, Any]: |
| return _request("POST", "/export", json=payload) |
|
|
