Spaces:
Runtime error
Runtime error
Download src/writing_studio/utils/monitoring.py from empirenexus/WritingStudio: direct link, hf CLI and curl.
- Browser
- Download file 3.17 kB
-
https://huggingface.co/spaces/empirenexus/WritingStudio/resolve/main/src/writing_studio/utils/monitoring.py
- Command line
-
hf download hf://spaces/empirenexus/WritingStudio/src/writing_studio/utils/monitoring.py
-
curl -L -o monitoring.py https://huggingface.co/spaces/empirenexus/WritingStudio/resolve/main/src/writing_studio/utils/monitoring.py
3.17 kB
| """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() | |