| from http.server import HTTPServer, BaseHTTPRequestHandler | |
| class HealthHandler(BaseHTTPRequestHandler): | |
| def do_GET(self): | |
| self.send_response(200) | |
| self.end_headers() | |
| self.wfile.write(b"OK") | |
| HTTPServer(("0.0.0.0", 7860), HealthHandler).serve_forever() | |
| # """ | |
| # Health Check Endpoint | |
| # Monitors system health and dependencies | |
| # """ | |
| # from datetime import datetime | |
| # from db import Database | |
| # from cache import cache | |
| # import os | |
| # def check_database(): | |
| # """Check if database is accessible""" | |
| # try: | |
| # db = Database() | |
| # # Try a simple operation | |
| # if db.client: | |
| # # Test query | |
| # db.client.table("users").select("*").limit(1).execute() | |
| # return True | |
| # return False | |
| # except Exception as e: | |
| # print(f"Database health check failed: {e}") | |
| # return False | |
| # def check_redis(): | |
| # """Check if Redis is accessible""" | |
| # try: | |
| # if not cache.enabled: | |
| # return False | |
| # # Try to set and get a test value | |
| # cache.set("health_check", "ok", ttl=10) | |
| # result = cache.get("health_check") | |
| # return result == "ok" | |
| # except Exception as e: | |
| # print(f"Redis health check failed: {e}") | |
| # return False | |
| # def check_livekit(): | |
| # """Check if LiveKit credentials are configured""" | |
| # return all([ | |
| # os.getenv("LIVEKIT_URL"), | |
| # os.getenv("LIVEKIT_API_KEY"), | |
| # os.getenv("LIVEKIT_API_SECRET") | |
| # ]) | |
| # def check_llm(): | |
| # """Check if LLM API keys are configured""" | |
| # return os.getenv("GROQ_API_KEY") is not None | |
| # def check_tts(): | |
| # """Check if TTS API keys are configured""" | |
| # return os.getenv("DEEPGRAM_API_KEY") is not None | |
| # def get_health_status(): | |
| # """Get comprehensive health status""" | |
| # checks = { | |
| # "database": check_database(), | |
| # "redis": check_redis(), | |
| # "livekit": check_livekit(), | |
| # "llm": check_llm(), | |
| # "tts": check_tts() | |
| # } | |
| # all_healthy = all(checks.values()) | |
| # return { | |
| # "status": "healthy" if all_healthy else "degraded", | |
| # "timestamp": datetime.now().isoformat(), | |
| # "checks": checks, | |
| # "version": "1.0.0" | |
| # } | |