Spaces:
Sleeping
Sleeping
Create health_check.py
Browse files- checks/health_check.py +38 -0
checks/health_check.py
ADDED
@@ -0,0 +1,38 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
1 |
+
# checks/health_check.py
|
2 |
+
import os
|
3 |
+
import requests
|
4 |
+
from typing import Dict, Tuple
|
5 |
+
|
6 |
+
def check_model_endpoint(endpoint_uri: str) -> Tuple[bool, Dict]:
|
7 |
+
"""Comprehensive check of model endpoint health"""
|
8 |
+
# Basic availability check
|
9 |
+
status_info = check_public_endpoint(endpoint_uri)
|
10 |
+
|
11 |
+
if not status_info["status"] or status_info["status_code"] in [503, 502]:
|
12 |
+
return False, status_info
|
13 |
+
|
14 |
+
# Test actual model response
|
15 |
+
try:
|
16 |
+
test_prompt = "Hello, how are you?"
|
17 |
+
response = requests.post(
|
18 |
+
endpoint_uri,
|
19 |
+
json={"inputs": test_prompt},
|
20 |
+
headers={"Authorization": f"Bearer {os.getenv('HF_TOKEN')}"},
|
21 |
+
timeout=10
|
22 |
+
)
|
23 |
+
response.raise_for_status()
|
24 |
+
return True, {
|
25 |
+
"status": True,
|
26 |
+
"status_code": response.status_code,
|
27 |
+
"response": response.json()
|
28 |
+
}
|
29 |
+
except Exception as e:
|
30 |
+
return False, {
|
31 |
+
"status": False,
|
32 |
+
"status_code": 500,
|
33 |
+
"error": str(e)
|
34 |
+
}
|
35 |
+
|
36 |
+
def should_launch_ui(status_info: Dict) -> bool:
|
37 |
+
"""Determine if UI should be launched based on status"""
|
38 |
+
return status_info.get("status", False) and status_info.get("status_code", 500) == 200
|