Coverage for src/lexigram/web/routing/health.py: 0%

1"""Health check route registration for lexigram-web.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, cast 

6 

7from starlette.responses import JSONResponse, Response 

8 

9from lexigram.contracts.core import HealthStatus 

10from lexigram.logging import get_logger 

11from lexigram.web import constants as const 

12from lexigram.web.routing.health_checks import WebHealthChecker 

13 

14if TYPE_CHECKING: 

15 from starlette.requests import Request 

16 

17 from lexigram.app.base import Application 

18 from lexigram.web.di.provider import WebProvider 

19 

20logger = get_logger(__name__) 

21 

22_STATUS_CODE_MAP: dict[HealthStatus, int] = { 

23 HealthStatus.HEALTHY: 200, 

24 HealthStatus.DEGRADED: 207, 

25 HealthStatus.UNHEALTHY: 503, 

26 HealthStatus.UNKNOWN: 503, 

27 HealthStatus.STARTING: 503, 

28} 

29 

30 

31async def _resolve_application(request: Request) -> Application: 

32 """Resolve the lexigram Application from the Starlette request.""" 

33 from lexigram.app.base import Application as LexigramApplication 

34 

35 container = getattr(getattr(request, "app", None), "state", None) 

36 container = getattr(container, "container", None) 

37 if container is None: 

38 raise RuntimeError("Application container not available") 

39 return cast( 

40 "LexigramApplication", 

41 await container.resolve(LexigramApplication, bypass_visibility=True), 

42 ) 

43 

44 

45def _status_code_for(status: HealthStatus) -> int: 

46 """Map a health status to an HTTP status code.""" 

47 return _STATUS_CODE_MAP.get(status, 503) 

48 

49 

50async def _probe_response(request: Request, probe_name: str) -> Response: 

51 """Run a named application probe and return a JSON response.""" 

52 app = await _resolve_application(request) 

53 probe = getattr(app, probe_name) 

54 result = await probe() 

55 return JSONResponse(result.to_dict(), status_code=_status_code_for(result.status)) 

56 

57 

58def register_health_route(provider: WebProvider, _app: Application) -> None: 

59 """Register comprehensive health check routes. 

60 

61 Idempotent: skips registration when a route for the health path is 

62 already present (e.g. registered before contributor mounts so that 

63 extension-provided routes cannot shadow the canonical endpoint). 

64 """ 

65 starlette = provider.starlette 

66 if starlette is None: 

67 raise RuntimeError("Starlette application not initialized") 

68 

69 prefix = str( 

70 getattr( 

71 provider.web_config, 

72 "health_check_prefix", 

73 const.DEFAULT_HEALTH_PATH, 

74 ) 

75 ) 

76 if any( 

77 getattr(route, "path", None) == prefix 

78 for route in getattr(starlette, "routes", []) 

79 ): 

80 logger.debug("web.health_route_already_registered", path=prefix) 

81 return 

82 

83 async def health_handler(request: Request) -> Response: 

84 """Return 200/207/503 based on real dependency health checks.""" 

85 container = getattr(getattr(request, "app", None), "state", {}) 

86 container = getattr(container, "container", None) 

87 

88 db_provider = None 

89 redis_client = None 

90 

91 if container is not None: 

92 try: 

93 from lexigram.contracts.data import DatabaseProviderProtocol 

94 

95 db_provider = await container.resolve(DatabaseProviderProtocol) 

96 except Exception: # noqa: BLE001, S110 

97 pass 

98 try: 

99 from lexigram.contracts.infra.cache import CacheBackendProtocol 

100 

101 redis_client = await container.resolve(CacheBackendProtocol) 

102 except Exception: # noqa: BLE001, S110 

103 pass 

104 

105 checker = WebHealthChecker( 

106 db_provider=db_provider, 

107 cache_backend=redis_client, 

108 ) 

109 result = await checker.check_health() 

110 status_code = _status_code_for(result.status) 

111 return JSONResponse(result.model_dump(mode="json"), status_code=status_code) 

112 

113 async def liveness_handler(request: Request) -> Response: 

114 """Return liveness probe results from the application.""" 

115 return await _probe_response(request, "liveness") 

116 

117 async def readiness_handler(request: Request) -> Response: 

118 """Return readiness probe results from the application.""" 

119 return await _probe_response(request, "readiness") 

120 

121 async def startup_handler(request: Request) -> Response: 

122 """Return startup probe results from the application.""" 

123 return await _probe_response(request, "startup_check") 

124 

125 # Use prefix from config if available 

126 starlette.add_route(prefix, health_handler, methods=["GET"]) 

127 starlette.add_route(f"{prefix.rstrip('/')}/live", liveness_handler, methods=["GET"]) 

128 starlette.add_route( 

129 f"{prefix.rstrip('/')}/ready", 

130 readiness_handler, 

131 methods=["GET"], 

132 ) 

133 starlette.add_route( 

134 f"{prefix.rstrip('/')}/startup", 

135 startup_handler, 

136 methods=["GET"], 

137 )