Spaces:
Sleeping
Sleeping
| """Health checks and monitoring utilities.""" | |
| import time | |
| from typing import Dict, Any | |
| from writing_studio.core.config import settings | |
| from writing_studio.services.model_service import get_model_service | |
| from writing_studio.utils.logging import logger | |
| class HealthCheck: | |
| """Health check service for monitoring application status.""" | |
| def __init__(self): | |
| """Initialize health check service.""" | |
| self.start_time = time.time() | |
| def check_health(self) -> Dict[str, Any]: | |
| """ | |
| Perform comprehensive health check. | |
| Returns: | |
| Health status dictionary | |
| """ | |
| status = { | |
| "status": "healthy", | |
| "timestamp": time.time(), | |
| "uptime_seconds": time.time() - self.start_time, | |
| "checks": {}, | |
| } | |
| # Check model service | |
| try: | |
| model_service = get_model_service() | |
| model_info = model_service.get_model_info() | |
| status["checks"]["model"] = { | |
| "status": "healthy", | |
| "details": model_info, | |
| } | |
| except Exception as e: | |
| logger.error(f"Model health check failed: {e}") | |
| status["checks"]["model"] = { | |
| "status": "unhealthy", | |
| "error": str(e), | |
| } | |
| status["status"] = "degraded" | |
| # Check configuration | |
| try: | |
| config_check = { | |
| "environment": settings.environment, | |
| "debug": settings.debug, | |
| "cache_enabled": settings.enable_cache, | |
| } | |
| status["checks"]["configuration"] = { | |
| "status": "healthy", | |
| "details": config_check, | |
| } | |
| except Exception as e: | |
| logger.error(f"Configuration check failed: {e}") | |
| status["checks"]["configuration"] = { | |
| "status": "unhealthy", | |
| "error": str(e), | |
| } | |
| status["status"] = "degraded" | |
| return status | |
| def check_readiness(self) -> Dict[str, Any]: | |
| """ | |
| Check if application is ready to serve requests. | |
| Returns: | |
| Readiness status dictionary | |
| """ | |
| try: | |
| # Ensure model is loaded | |
| model_service = get_model_service() | |
| if model_service._current_model is None: | |
| return { | |
| "ready": False, | |
| "reason": "Model not loaded", | |
| } | |
| return { | |
| "ready": True, | |
| "timestamp": time.time(), | |
| } | |
| except Exception as e: | |
| logger.error(f"Readiness check failed: {e}") | |
| return { | |
| "ready": False, | |
| "reason": str(e), | |
| } | |
| def check_liveness(self) -> Dict[str, Any]: | |
| """ | |
| Check if application is alive. | |
| Returns: | |
| Liveness status dictionary | |
| """ | |
| return { | |
| "alive": True, | |
| "timestamp": time.time(), | |
| "uptime_seconds": time.time() - self.start_time, | |
| } | |
| # Global health check instance | |
| health_check = HealthCheck() | |