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

1"""Health check utilities and endpoints. 

2 

3All health endpoints are public (no authentication required): 

4- GET /health/ - Comprehensive health status 

5- GET /health/live - Kubernetes liveness probe 

6- GET /health/ready - Kubernetes readiness probe 

7 

8These endpoints are configured as public in Kong API Gateway. 

9""" 

10 

11from datetime import UTC, datetime 

12from typing import Any 

13 

14from fastapi import APIRouter, Depends 

15 

16from . import checker 

17from .checker import HealthStatus 

18from .schemas import HealthResponse 

19 

20 

21def get_health_checker() -> HealthStatus: 

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

23 return checker.get_health_checker() 

24 

25 

26async def basic_health_check(): 

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

28 return ( 

29 "healthy", 

30 "Service is running", 

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

32 ) 

33 

34 

35async def database_health_check(): 

36 """Check database connectivity.""" 

37 try: 

38 # This would check database connection 

39 # For now, just return healthy 

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

41 except Exception as e: 

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

43 

44 

45def create_health_router( 

46 service_name: str, service_version: str, include_database_check: bool = True 

47) -> APIRouter: 

48 """Create health check router with standard endpoints. 

49 

50 Args: 

51 service_name: Name of the service 

52 service_version: Version of the service 

53 include_database_check: Whether to include database health check 

54 

55 Returns: 

56 APIRouter with health check endpoints 

57 """ 

58 # Set the global _health_checker in checker module 

59 checker._health_checker = HealthStatus(service_name, service_version) 

60 

61 router = APIRouter(prefix="/health", tags=["Health"]) 

62 

63 # Initialize health checker 

64 health_checker = checker._health_checker 

65 

66 # Add basic checks 

67 health_checker.add_check("basic", basic_health_check, critical=True) 

68 

69 if include_database_check: 

70 health_checker.add_check("database", database_health_check, critical=True) 

71 

72 @router.get("/", response_model=HealthResponse) 

73 async def health_check( 

74 checker: HealthStatus = Depends(get_health_checker), # noqa: B008 

75 ) -> HealthResponse: 

76 """Get comprehensive health status.""" 

77 health_response = await checker.get_health() 

78 return health_response 

79 

80 @router.get("/live") 

81 async def liveness_probe(): 

82 """Kubernetes liveness probe endpoint.""" 

83 return {"status": "alive", "timestamp": datetime.now(UTC)} 

84 

85 @router.get("/ready") 

86 async def readiness_probe( 

87 checker: HealthStatus = Depends(get_health_checker), # noqa: B008 

88 ) -> dict[str, Any]: 

89 """Kubernetes readiness probe endpoint.""" 

90 health_response = await checker.get_health() 

91 

92 if health_response.status == "healthy": 

93 return {"status": "ready", "timestamp": datetime.now(UTC)} 

94 else: 

95 return {"status": "not_ready", "timestamp": datetime.now(UTC)} 

96 

97 return router