You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

45 lines
1.5 KiB

from django.db import connection
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
import os
import redis
class HealthCheckView(APIView):
"""
Endpoint to check system health (database connection and redis connection).
"""
authentication_classes = []
permission_classes = []
def get(self, request):
db_status = "healthy"
redis_status = "healthy"
# Check Database
try:
connection.ensure_connection()
except Exception as e:
db_status = f"unhealthy: {str(e)}"
# Check Redis
from django.conf import settings
is_eager = getattr(settings, 'CELERY_TASK_ALWAYS_EAGER', False)
try:
redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0")
r = redis.Redis.from_url(redis_url, socket_timeout=3)
r.ping()
except Exception as e:
if is_eager:
redis_status = "healthy" # Marked as healthy since Redis is optional/unused in eager mode
else:
redis_status = f"unhealthy: {str(e)}"
is_healthy = db_status == "healthy" and redis_status == "healthy"
status_code = status.HTTP_200_OK if is_healthy else status.HTTP_503_SERVICE_UNAVAILABLE
return Response({
"status": "healthy" if is_healthy else "unhealthy",
"database": db_status,
"redis": redis_status
}, status=status_code)