Coverage for src / lexigram / contracts / core / health.py: 28%

1"""Health check types for Lexigram Framework.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from datetime import datetime 

7from enum import StrEnum 

8from typing import Any, Protocol, runtime_checkable 

9 

10 

11class HealthCheckCategory(StrEnum): 

12 """Categorises health checks by their role in the Kubernetes health model. 

13 

14 Attributes: 

15 LIVENESS: Checks that the application is alive (not deadlocked). 

16 Used by ``/health/live`` endpoints. Failure triggers a restart. 

17 READINESS: Checks that the application is ready to accept traffic. 

18 Used by ``/health/ready`` endpoints. Failure removes the instance 

19 from the load-balancer rotation. 

20 STARTUP: Checks that the application has finished its startup sequence. 

21 Used by ``/health/startup`` endpoints. Succeeded once; after that 

22 the probe switches to liveness/readiness. 

23 """ 

24 

25 LIVENESS = "liveness" 

26 READINESS = "readiness" 

27 STARTUP = "startup" 

28 

29 

30class HealthStatus(StrEnum): 

31 """Unified health status.""" 

32 

33 HEALTHY = "healthy" 

34 UNHEALTHY = "unhealthy" 

35 DEGRADED = "degraded" 

36 UNKNOWN = "unknown" 

37 STARTING = "starting" 

38 

39 

40@runtime_checkable 

41class HealthCheckProtocol(Protocol): 

42 """Protocol for health check capability. 

43 

44 Expected Behavior: 

45 - health_check: MUST return a result within a reasonable timeout (~5s). 

46 - health_check: SHOULD NOT raise exceptions; wrap errors in HealthCheckResult. 

47 """ 

48 

49 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: ... 

50 

51 

52@dataclass(frozen=True) 

53class HealthCheckResult: 

54 """Result of a health check.""" 

55 

56 component: str 

57 status: HealthStatus = HealthStatus.HEALTHY 

58 message: str | None = None 

59 error: str | None = None 

60 duration_ms: float = 0.0 

61 details: dict[str, Any] | None = None 

62 checked_at: datetime | None = None 

63 category: HealthCheckCategory = HealthCheckCategory.READINESS 

64 

65 def to_dict(self) -> dict[str, Any]: 

66 """Convert to dictionary format.""" 

67 result: dict[str, Any] = { 

68 "component": self.component, 

69 "category": self.category.value, 

70 "status": str(self.status), 

71 "duration_ms": round(self.duration_ms, 2), 

72 } 

73 if self.message: 

74 result["message"] = self.message 

75 if self.error: 

76 result["error"] = self.error 

77 if self.details: 

78 result["details"] = self.details 

79 if self.checked_at: 

80 result["checked_at"] = self.checked_at.isoformat() 

81 return result 

82 

83 def is_healthy(self) -> bool: 

84 """Check if status is healthy.""" 

85 return self.status == HealthStatus.HEALTHY 

86 

87 def is_degraded(self) -> bool: 

88 """Check if status is degraded.""" 

89 return self.status == HealthStatus.DEGRADED 

90 

91 

92@dataclass(frozen=True) 

93class AggregateHealthResult: 

94 """Composite health result that collects checks from multiple components. 

95 

96 The aggregate status follows a ``worst-case`` rule: 

97 * Any ``UNHEALTHY`` check → overall ``UNHEALTHY`` 

98 * Any ``DEGRADED`` check (no UNHEALTHY) → overall ``DEGRADED`` 

99 * All ``HEALTHY`` → overall ``HEALTHY`` 

100 * No checks registered → ``UNKNOWN`` 

101 """ 

102 

103 components: list[HealthCheckResult] = field(default_factory=list) 

104 

105 @property 

106 def status(self) -> HealthStatus: 

107 """Overall status, computed from component checks.""" 

108 if not self.components: 

109 return HealthStatus.UNKNOWN 

110 if any(c.status == HealthStatus.UNHEALTHY for c in self.components): 

111 return HealthStatus.UNHEALTHY 

112 if any(c.status == HealthStatus.DEGRADED for c in self.components): 

113 return HealthStatus.DEGRADED 

114 return HealthStatus.HEALTHY 

115 

116 def is_healthy(self) -> bool: 

117 """Return True only if all components are healthy.""" 

118 return self.status == HealthStatus.HEALTHY 

119 

120 def to_dict(self) -> dict[str, Any]: 

121 """Convert to a dictionary suitable for a ``/health`` HTTP response.""" 

122 return { 

123 "status": self.status.value, 

124 "components": [c.to_dict() for c in self.components], 

125 } 

126 

127 

128@runtime_checkable 

129class HealthCheckAggregatorProtocol(Protocol): 

130 """Protocol for services that aggregate health checks from multiple providers. 

131 

132 Implementations collect :class:`HealthCheckProtocol` instances, run them 

133 concurrently, and return a single :class:`AggregateHealthResult`. 

134 """ 

135 

136 def register( 

137 self, 

138 name: str, 

139 check: HealthCheckProtocol, 

140 *, 

141 category: HealthCheckCategory = HealthCheckCategory.READINESS, 

142 ) -> None: 

143 """Register a named health check. 

144 

145 Args: 

146 name: Unique component name for this check. 

147 check: Object implementing :class:`HealthCheckProtocol`. 

148 category: Probe category for this check. Defaults to 

149 :class:`HealthCheckCategory.READINESS`. 

150 """ 

151 ... 

152 

153 async def run_all( 

154 self, 

155 timeout: float = 5.0, 

156 *, 

157 category: HealthCheckCategory | None = None, 

158 ) -> AggregateHealthResult: 

159 """Run all registered checks concurrently and aggregate results. 

160 

161 Args: 

162 timeout: Per-check timeout in seconds. 

163 category: Optional probe category filter. 

164 

165 Returns: 

166 :class:`AggregateHealthResult` containing all component results. 

167 """ 

168 ... 

169 

170 async def run_liveness(self, timeout: float = 5.0) -> AggregateHealthResult: 

171 """Run only liveness checks and aggregate results.""" 

172 ... 

173 

174 async def run_readiness(self, timeout: float = 5.0) -> AggregateHealthResult: 

175 """Run only readiness checks and aggregate results.""" 

176 ... 

177 

178 async def run_startup(self, timeout: float = 5.0) -> AggregateHealthResult: 

179 """Run only startup checks and aggregate results.""" 

180 ... 

181 

182 

183# Canonical alias — infrastructure protocols should reference this name. 

184# Identical to HealthCheckProtocol; provided as a mixin-friendly import. 

185HealthCheckableProtocol = HealthCheckProtocol 

186 

187__all__ = [ 

188 "AggregateHealthResult", 

189 "HealthCheckAggregatorProtocol", 

190 "HealthCheckCategory", 

191 "HealthCheckProtocol", 

192 "HealthCheckResult", 

193 "HealthCheckableProtocol", 

194 "HealthStatus", 

195]