Coverage for src / mysingle / core / health / checker.py: 0%

1"""Health check utilities and endpoints.""" 

2 

3from collections.abc import Callable 

4from datetime import UTC, datetime 

5 

6from .schemas import HealthResponse 

7 

8 

9class HealthStatus: 

10 """Health status checker with configurable checks.""" 

11 

12 def __init__(self, service_name: str, service_version: str): 

13 self.service_name = service_name 

14 self.service_version = service_version 

15 self.start_time = datetime.now(UTC) 

16 self.checks: dict[str, dict] = {} 

17 

18 def add_check(self, name: str, check_func: Callable, critical: bool = True) -> None: 

19 """Add a health check function. 

20 

21 Args: 

22 name: Name of the check 

23 check_func: Async function that returns (status, message, details) 

24 critical: Whether this check is critical for overall health 

25 """ 

26 self.checks[name] = {"func": check_func, "critical": critical} 

27 

28 async def get_health(self) -> HealthResponse: 

29 """Get comprehensive health status.""" 

30 now = datetime.now(UTC) 

31 uptime = (now - self.start_time).total_seconds() 

32 

33 check_results = {} 

34 overall_status = "healthy" 

35 

36 # Run all health checks 

37 for check_name, check_config in self.checks.items(): 

38 try: 

39 status_result, message, details = await check_config["func"]() 

40 check_results[check_name] = { 

41 "status": status_result, 

42 "message": message, 

43 "details": details, 

44 "critical": check_config["critical"], 

45 } 

46 

47 # Update overall status if critical check fails 

48 if check_config["critical"] and status_result != "healthy": 

49 overall_status = "unhealthy" 

50 

51 except Exception as e: 

52 check_results[check_name] = { 

53 "status": "error", 

54 "message": f"Check failed: {str(e)}", 

55 "details": {}, 

56 "critical": check_config["critical"], 

57 } 

58 

59 if check_config["critical"]: 

60 overall_status = "unhealthy" 

61 

62 return HealthResponse( 

63 status=overall_status, 

64 timestamp=now, 

65 service=self.service_name, 

66 version=self.service_version, 

67 uptime=uptime, 

68 checks=check_results, 

69 ) 

70 

71 

72# Global health checker instance 

73_health_checker: HealthStatus | None = None 

74 

75 

76def get_health_checker() -> HealthStatus: 

77 """Get the global health checker instance.""" 

78 global _health_checker 

79 if _health_checker is None: 

80 raise RuntimeError( 

81 "Health checker not initialized. Call create_health_router first." 

82 ) 

83 return _health_checker 

84 

85 

86async def basic_health_check(): 

87 """Basic health check that always passes.""" 

88 return ( 

89 "healthy", 

90 "Service is running", 

91 {"timestamp": datetime.now(UTC).isoformat()}, 

92 ) 

93 

94 

95async def database_health_check(): 

96 """Check database connectivity.""" 

97 try: 

98 # This would check database connection 

99 # For now, just return healthy 

100 return "healthy", "Database connection OK", {"connection": "active"} 

101 except Exception as e: 

102 return "unhealthy", f"Database error: {str(e)}", {"error": str(e)}