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 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: 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)