Spaces:
Configuration error
Configuration error
| """FormFav API client. | |
| Server-side only. Uses X-API-Key header (NOT Authorization: Bearer). | |
| Handles retry/backoff, timeouts, caching, and graceful degradation when | |
| premium fields (e.g. /predictions) are unavailable. | |
| """ | |
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| import logging | |
| import time | |
| from pathlib import Path | |
| from typing import Any, Optional | |
| import httpx | |
| from ..config import settings | |
| logger = logging.getLogger("trifecta_bro.api") | |
| class FormFavError(Exception): | |
| """Base error for FormFav interactions.""" | |
| class FormFavAuthError(FormFavError): | |
| """401/403 — key missing or lacks entitlement.""" | |
| class FormFavNotFoundError(FormFavError): | |
| """404 — invalid meeting/race params.""" | |
| class FormFavClient: | |
| def __init__( | |
| self, | |
| api_key: Optional[str] = None, | |
| base_url: Optional[str] = None, | |
| cache_dir: Optional[Path] = None, | |
| cache_ttl: Optional[int] = None, | |
| timeout: Optional[float] = None, | |
| max_retries: Optional[int] = None, | |
| ) -> None: | |
| self.api_key = api_key if api_key is not None else settings.api_key | |
| self.base_url = (base_url or settings.base_url).rstrip("/") | |
| self.cache_dir = Path(cache_dir or settings.cache_dir) | |
| self.cache_ttl = cache_ttl if cache_ttl is not None else settings.cache_ttl_seconds | |
| self.timeout = timeout if timeout is not None else settings.http_timeout | |
| self.max_retries = max_retries if max_retries is not None else settings.max_retries | |
| self.cache_dir.mkdir(parents=True, exist_ok=True) | |
| # ----- low level ----- | |
| def _headers(self) -> dict[str, str]: | |
| if not self.api_key: | |
| raise FormFavAuthError("FORMFAV_API_KEY is not set in the environment.") | |
| return {"X-API-Key": self.api_key, "Accept": "application/json"} | |
| def _cache_path(self, key: str) -> Path: | |
| h = hashlib.sha1(key.encode()).hexdigest()[:16] | |
| return self.cache_dir / f"{h}.json" | |
| def _cache_get(self, key: str) -> Optional[Any]: | |
| p = self._cache_path(key) | |
| if not p.exists(): | |
| return None | |
| try: | |
| if time.time() - p.stat().st_mtime > self.cache_ttl: | |
| return None | |
| return json.loads(p.read_text()) | |
| except Exception as exc: # pragma: no cover - defensive | |
| logger.warning("cache read failed: %s", exc) | |
| return None | |
| def _cache_set(self, key: str, value: Any) -> None: | |
| try: | |
| self._cache_path(key).write_text(json.dumps(value)) | |
| except Exception as exc: # pragma: no cover - defensive | |
| logger.warning("cache write failed: %s", exc) | |
| def _request(self, path: str, params: dict[str, Any]) -> Any: | |
| url = f"{self.base_url}{path}" | |
| cache_key = f"{path}?{sorted((k, str(v)) for k, v in params.items())}" | |
| cached = self._cache_get(cache_key) | |
| if cached is not None: | |
| logger.info("cache hit: %s", path) | |
| return cached | |
| last_exc: Optional[Exception] = None | |
| for attempt in range(1, self.max_retries + 1): | |
| try: | |
| with httpx.Client(timeout=self.timeout) as client: | |
| resp = client.get(url, headers=self._headers(), params=params) | |
| if resp.status_code == 401: | |
| raise FormFavAuthError("401 Unauthorized — check FORMFAV_API_KEY") | |
| if resp.status_code == 403: | |
| raise FormFavAuthError("403 Forbidden — key lacks entitlement for this endpoint") | |
| if resp.status_code == 404: | |
| raise FormFavNotFoundError(f"404 Not Found: {url} params={params}") | |
| resp.raise_for_status() | |
| data = resp.json() | |
| self._cache_set(cache_key, data) | |
| return data | |
| except (FormFavAuthError, FormFavNotFoundError): | |
| raise | |
| except Exception as exc: # noqa: BLE001 - retry transient errors | |
| last_exc = exc | |
| wait = min(2 ** attempt, 30) | |
| logger.warning("request attempt %s failed: %s (retry in %ss)", attempt, exc, wait) | |
| if attempt < self.max_retries: | |
| time.sleep(wait) | |
| raise FormFavError(f"FormFav request failed after {self.max_retries} attempts: {last_exc}") | |
| # ----- endpoints ----- | |
| def get_meetings(self, date: str, country: Optional[str] = None, | |
| race_code: Optional[str] = None) -> list[dict[str, Any]]: | |
| params = {"date": date} | |
| if country or settings.country: | |
| params["country"] = country or settings.country | |
| if race_code or settings.race_code: | |
| params["race_code"] = race_code or settings.race_code | |
| data = self._request("/form/meetings", params) | |
| return data.get("meetings", []) | |
| def get_race_form(self, date: str, track: str, race: int | str, | |
| country: Optional[str] = None, | |
| race_code: Optional[str] = None, | |
| timezone: Optional[str] = None) -> dict[str, Any]: | |
| params = { | |
| "date": date, | |
| "track": track, | |
| "race": race, | |
| "country": country or settings.country, | |
| "race_code": race_code or settings.race_code, | |
| } | |
| if timezone or settings.default_timezone: | |
| params["timezone"] = timezone or settings.default_timezone | |
| return self._request("/form", params) | |
| def get_race_predictions(self, date: str, track: str, race: int | str, | |
| country: Optional[str] = None, | |
| race_code: Optional[str] = None) -> Optional[dict[str, Any]]: | |
| """Premium endpoint. Returns None when the key lacks entitlement.""" | |
| params = { | |
| "date": date, "track": track, "race": race, | |
| "country": country or settings.country, | |
| "race_code": race_code or settings.race_code, | |
| } | |
| try: | |
| return self._request("/predictions", params) | |
| except FormFavAuthError: | |
| logger.info("/predictions unavailable (403) — falling back to local model only") | |
| return None | |