Spaces:
Runtime error
Runtime error
File size: 3,314 Bytes
18ade12 1c7cfea 18ade12 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | """
Health check API endpoints.
Provides endpoints for checking application and database health:
- GET /health: Overall application health
- GET /health/db: Database-only health
Per contracts/health-endpoint.yaml OpenAPI specification.
"""
from datetime import datetime, timezone
from typing import Union
from fastapi import APIRouter, Response, status
from src.api.schemas.health import (
DatabaseHealthResponse,
HealthChecks,
HealthResponse,
)
from src.db.health import check_database_health, get_full_health_status
router = APIRouter(prefix="/health", tags=["Health"])
# Default timeout for health checks (10 seconds for Neon cold starts)
HEALTH_CHECK_TIMEOUT = 10.0
@router.get(
"",
response_model=HealthResponse,
summary="Application Health Check",
description="Returns the health status of the application including database connectivity. "
"This endpoint does not require authentication.",
responses={
200: {
"description": "Health check successful",
"model": HealthResponse,
},
503: {
"description": "Service unhealthy",
"model": HealthResponse,
},
},
)
async def get_health(response: Response) -> HealthResponse:
"""
Get overall application health status.
Returns 200 if healthy, 503 if any component is unhealthy.
"""
health_data = await get_full_health_status(timeout_seconds=HEALTH_CHECK_TIMEOUT)
# Set appropriate HTTP status code
if health_data["status"] == "unhealthy":
response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
# Convert to response model
db_health = health_data["checks"]["database"]
return HealthResponse(
status=health_data["status"],
timestamp=datetime.fromisoformat(health_data["timestamp"].replace("Z", "+00:00")),
checks=HealthChecks(
database=DatabaseHealthResponse(
status=db_health["status"],
connected=db_health["connected"],
latency_ms=db_health.get("latency_ms"),
error=db_health.get("error"),
)
),
)
@router.get(
"/db",
response_model=DatabaseHealthResponse,
summary="Database-Only Health Check",
description="Returns only the database connectivity status. "
"Useful for targeted monitoring of database health. "
"This endpoint does not require authentication.",
responses={
200: {
"description": "Database is healthy",
"model": DatabaseHealthResponse,
},
503: {
"description": "Database is unhealthy",
"model": DatabaseHealthResponse,
},
},
)
async def get_database_health(response: Response) -> DatabaseHealthResponse:
"""
Get database-only health status.
Returns 200 if database is healthy, 503 if unhealthy.
"""
db_health = await check_database_health(timeout_seconds=HEALTH_CHECK_TIMEOUT)
# Set appropriate HTTP status code
if db_health.status == "unhealthy":
response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
return DatabaseHealthResponse(
status=db_health.status,
connected=db_health.connected,
latency_ms=db_health.latency_ms,
error=db_health.error,
)
|