Spaces:
Configuration error
Configuration error
File size: 6,269 Bytes
e23172f | 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 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 | """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
|