Coverage for src/lexigram/web/routing/health_checks.py: 33%
85 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""Health checking utilities separated from route handlers.
3This module contains the HealthChecker and related Pydantic models. Keeping
4this logic here helps keep `health.py` small and focused on HTTP handlers.
5"""
7from __future__ import annotations
9import asyncio
10from dataclasses import dataclass
11from datetime import datetime
12from typing import TYPE_CHECKING, cast
14from lexigram.contracts.core import HealthStatus
15from lexigram.domain import DomainModel
16from lexigram.logging import get_logger
17from lexigram.primitives import clock as ambient_clock
19if TYPE_CHECKING:
20 from lexigram.contracts.data import DatabaseProviderProtocol
21 from lexigram.contracts.infra.cache import CacheBackendProtocol
23logger = get_logger(__name__)
26@dataclass(init=False)
27class ComponentHealth(DomainModel):
28 status: HealthStatus
29 latency_ms: float
30 checked_at: datetime
31 message: str | None = None
34@dataclass(init=False)
35class HealthCheckResponse(DomainModel):
36 status: HealthStatus
37 version: str
38 components: dict[str, ComponentHealth]
39 checked_at: datetime
42class WebHealthChecker:
43 """Health checker for web application dependencies.
45 Not to be confused with ``lexigram.health.HealthChecker`` from core,
46 which is the general-purpose health check aggregator.
47 """
49 def __init__(
50 self,
51 *,
52 db_provider: DatabaseProviderProtocol | None = None,
53 cache_backend: CacheBackendProtocol | None = None,
54 app_version: str = "unknown",
55 ) -> None:
56 self.db_provider = db_provider
57 self.cache_backend = cache_backend
58 self.app_version = app_version
60 async def check_health(self) -> HealthCheckResponse:
61 checked_at = ambient_clock.now()
63 db_health, redis_health = await asyncio.gather(
64 self._check_database(),
65 self._check_redis(),
66 return_exceptions=True,
67 )
69 if isinstance(db_health, Exception):
70 db_health = ComponentHealth(
71 status=HealthStatus.UNHEALTHY,
72 latency_ms=0,
73 message=f"Error: {db_health!s}",
74 checked_at=ambient_clock.now(),
75 )
77 if isinstance(redis_health, Exception):
78 redis_health = ComponentHealth(
79 status=HealthStatus.UNHEALTHY,
80 latency_ms=0,
81 message=f"Error: {redis_health!s}",
82 checked_at=ambient_clock.now(),
83 )
85 components = {"database": db_health, "redis": redis_health}
87 # At runtime exceptions are converted to ComponentHealth instances above.
88 # Use typing.cast to inform mypy that this dict contains ComponentHealth values.
89 components_typed = cast("dict[str, ComponentHealth]", components)
91 overall_status = self._determine_overall_status(components_typed)
93 return HealthCheckResponse(
94 status=overall_status,
95 version=self.app_version,
96 components=components_typed,
97 checked_at=checked_at,
98 )
100 async def _check_database(self) -> ComponentHealth:
101 if not self.db_provider:
102 return ComponentHealth(
103 status=HealthStatus.UNHEALTHY,
104 latency_ms=0,
105 message="Database provider not configured",
106 checked_at=ambient_clock.now(),
107 )
109 start = ambient_clock.now()
111 try:
112 # Use the contract's health_check method
113 result = await self.db_provider.health_check()
115 latency = (ambient_clock.now() - start).total_seconds() * 1000
117 # Map HealthCheckResult to ComponentHealth
118 if result.status == HealthStatus.HEALTHY:
119 status = HealthStatus.HEALTHY
120 message = result.message or "Database connection OK"
121 elif result.status == HealthStatus.DEGRADED:
122 status = HealthStatus.DEGRADED
123 message = result.message or "Database degraded"
124 else:
125 status = HealthStatus.UNHEALTHY
126 message = result.message or "Database unhealthy"
128 return ComponentHealth(
129 status=status,
130 latency_ms=latency,
131 message=message,
132 checked_at=ambient_clock.now(),
133 )
135 except Exception as exc: # noqa: BLE001 — health checks must not propagate; any error becomes an UNHEALTHY status
136 latency = (ambient_clock.now() - start).total_seconds() * 1000
137 logger.exception("Database health check failed")
138 return ComponentHealth(
139 status=HealthStatus.UNHEALTHY,
140 latency_ms=latency,
141 message=str(exc),
142 checked_at=ambient_clock.now(),
143 )
145 async def _check_redis(self) -> ComponentHealth:
146 if not self.cache_backend:
147 return ComponentHealth(
148 status=HealthStatus.UNHEALTHY,
149 latency_ms=0,
150 message="Cache backend not configured",
151 checked_at=ambient_clock.now(),
152 )
154 start = ambient_clock.now()
156 try:
157 # Use the contract's health_check method
158 result = await self.cache_backend.health_check()
160 latency = (ambient_clock.now() - start).total_seconds() * 1000
162 # Map HealthCheckResult to ComponentHealth
163 if result.status == HealthStatus.HEALTHY:
164 status = HealthStatus.HEALTHY
165 message = result.message or "Cache connection OK"
166 elif result.status == HealthStatus.DEGRADED:
167 status = HealthStatus.DEGRADED
168 message = result.message or "Cache degraded"
169 else:
170 status = HealthStatus.UNHEALTHY
171 message = result.message or "Cache unhealthy"
173 return ComponentHealth(
174 status=status,
175 latency_ms=latency,
176 message=message,
177 checked_at=ambient_clock.now(),
178 )
180 except Exception as exc: # noqa: BLE001 — health checks must not propagate; any error becomes an UNHEALTHY status
181 latency = (ambient_clock.now() - start).total_seconds() * 1000
182 logger.exception("Cache health check failed")
183 return ComponentHealth(
184 status=HealthStatus.UNHEALTHY,
185 latency_ms=latency,
186 message=str(exc),
187 checked_at=ambient_clock.now(),
188 )
190 def _determine_overall_status(
191 self,
192 components: dict[str, ComponentHealth],
193 ) -> HealthStatus:
194 statuses = [c.status for c in components.values()]
196 if HealthStatus.UNHEALTHY in statuses:
197 return HealthStatus.UNHEALTHY
198 if HealthStatus.DEGRADED in statuses:
199 return HealthStatus.DEGRADED
200 return HealthStatus.HEALTHY