52 lines
1.4 KiB
Python
52 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
import redis
|
|
from fastapi import APIRouter, status
|
|
|
|
from app.core.config import settings
|
|
from app.core.database import check_database_connection
|
|
from app.schemas.common import HealthResponse
|
|
|
|
router = APIRouter(tags=["Health"])
|
|
|
|
|
|
@router.get(
|
|
"/health",
|
|
response_model=HealthResponse,
|
|
status_code=status.HTTP_200_OK,
|
|
summary="Health Check",
|
|
description="Check the health status of the application and its dependencies.",
|
|
)
|
|
async def health_check() -> HealthResponse:
|
|
"""Perform health check on all system components."""
|
|
# Check database
|
|
db_status = "healthy" if check_database_connection() else "unhealthy"
|
|
|
|
# Check Redis
|
|
redis_status = "healthy"
|
|
try:
|
|
r = redis.Redis(
|
|
host=settings.redis_host,
|
|
port=settings.redis_port,
|
|
db=settings.redis_db,
|
|
password=settings.redis_password or None,
|
|
socket_timeout=3,
|
|
)
|
|
r.ping()
|
|
r.close()
|
|
except Exception:
|
|
redis_status = "unhealthy"
|
|
|
|
overall_status = "healthy" if db_status == "healthy" and redis_status == "healthy" else "degraded"
|
|
|
|
return HealthResponse(
|
|
status=overall_status,
|
|
version=settings.app_version,
|
|
environment=settings.app_env,
|
|
database=db_status,
|
|
redis=redis_status,
|
|
timestamp=datetime.utcnow(),
|
|
)
|