Spaces:
Sleeping
Sleeping
File size: 1,747 Bytes
6bf9c98 36334a9 6bf9c98 26127a9 36334a9 cdd1f75 074f0cf 26127a9 6bf9c98 26127a9 36334a9 074f0cf 6bf9c98 074f0cf 6bf9c98 074f0cf |
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 |
# status_check.py
from typing import Optional, Tuple
import os
import requests
from urllib.parse import urlparse
from endpoint_utils import hf_headers # or same module where you defined it
def resolve_endpoint() -> Optional[str]:
uri = (os.environ.get("HF_ENDPOINT_URI") or "").strip()
if not uri:
return None
p = urlparse(uri)
if p.scheme not in {"http", "https"} or not p.netloc:
# treat bad values like “localhost:8000” (no scheme) as missing
return None
return uri
def is_endpoint_healthy(uri: Optional[str], timeout: float = 5.0) -> Tuple[bool, str]:
if not uri:
return False, "no URI configured"
headers = hf_headers()
# Try /health first (some endpoints support it)
try:
r = requests.get(f"{uri.rstrip('/')}/health", headers=headers, timeout=timeout)
if r.ok:
return True, "OK"
try:
msg = (r.json().get("error") or r.json().get("message")) # type: ignore
except Exception:
msg = (r.text or "").strip()
if r.status_code in (401, 403):
return False, f"HTTP {r.status_code} – Unauthorized (check HF_TOKEN). {msg}"
except requests.RequestException:
pass
# Fallback: POST /
try:
r = requests.post(uri, headers=headers, json={"inputs": "ping"}, timeout=timeout)
if r.ok:
return True, "OK"
try:
msg = (r.json().get("error") or r.json().get("message")) # type: ignore
except Exception:
msg = (r.text or "").strip()
return False, f"HTTP {r.status_code}" + (f" – {msg}" if msg else "")
except requests.RequestException as e:
return False, type(e).__name__
|