Add healthcheck

api/urls.py

@@ -1,12 +1,14 @@
 from django.urls import path, include
 from rest_framework import routers
 
-from api.views import ServiceViewSet
+from api.views import ServiceViewSet, HealthCheckViewSet, PingViewSet
 
 app_name = "api"
 
 router = routers.DefaultRouter()
 router.register(r"services", ServiceViewSet, basename="service")
+router.register(r"health", HealthCheckViewSet, basename="health")
+router.register(r"ping", PingViewSet, basename="ping")
 
 # Wire up our API using automatic URL routing.
 # Additionally, we include login URLs for the browsable API.

api/views.py

@@ -6,10 +6,13 @@
 from django.core.files.base import ContentFile
 from django.urls import reverse
 from rest_framework import status
+from rest_framework.permissions import IsAdminUser
 from rest_framework.response import Response
 from rest_framework.viewsets import ModelViewSet
+from rest_framework.viewsets import ViewSet
 
 from api.serializers import ServiceSerializer
+from health.registry import HealthCheckRegistry
 from registry.models import (
     Service,
     ServiceURL,
@@ -142,3 +145,28 @@ def download_and_save_image(self, service, image_url):
         except requests.RequestException:
             service.image = None
             service.save()
+
+
+class HealthCheckViewSet(ViewSet):
+    """
+    ViewSet for health check API.
+    """
+
+    # Staff only
+    permission_classes = [IsAdminUser]
+
+    def list(self, request):
+        healthchecks = HealthCheckRegistry.get_registered_healthchecks()
+        results = {check.name: check.check(request=request) for check in healthchecks}
+        return Response(results)
+
+
+class PingViewSet(ViewSet):
+    """
+    ViewSet for ping API.
+    """
+
+    permission_classes = []
+
+    def list(self, request):
+        return Response({"status": "ok"})

data410/settings.py

@@ -60,6 +60,7 @@
     "api",
     "registry",
     "metrics",
+    "health",
     # Third Party Apps
     "storages",
     "rest_framework",

data410/urls.py

@@ -32,6 +32,7 @@
     path("registry/", include("registry.urls")),
     path("", include("core.urls")),
     path("api/", include("api.urls")),
+    path("health/", include("health.urls")),
     path(
         "sitemap.xml",
         sitemap,

health/admin.py

@@ -0,0 +1,3 @@
+from django.contrib import admin
+
+# Register your models here.

health/apps.py

@@ -0,0 +1,10 @@
+from django.apps import AppConfig
+
+
+class HealthConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "health"
+
+    def ready(self):
+        # Import the module containing healthchecks to trigger registration
+        import health.healthchecks.common  # noqa

health/decorators.py

@@ -0,0 +1,9 @@
+from health.registry import HealthCheckRegistry
+
+
+def register_healthcheck(healthcheck_class):
+    """
+    Decorator to register health checks automatically.
+    """
+    HealthCheckRegistry.register(healthcheck_class)
+    return healthcheck_class

health/healthchecks/base.py

@@ -0,0 +1,25 @@
+from typing import Dict, Any
+
+
+class BaseHealthCheck:
+    """
+    A base class for defining health checks.
+    Subclasses should implement the `check` method to perform specific health checks.
+    """
+
+    name: str = "base"
+
+    def check(self, **kwargs) -> Dict[str, Any]:
+        """
+        Perform the health check and return the status and additional details.
+        Must return a dictionary with at least `status: "ok" | "error"`.
+        """
+        raise NotImplementedError("Subclasses must implement the `check` method.")
+
+    @property
+    def verbose_name(self) -> str:
+        """
+        Return a human-readable name for the health check.
+        """
+        # Convert the name to title case
+        return self.name.replace("_", " ").title()

health/healthchecks/common.py

@@ -0,0 +1,84 @@
+import requests
+from django.core.cache import cache
+from django.core.files.base import ContentFile
+from django.core.files.storage import default_storage
+from django.db import connections
+from django.urls import reverse
+
+from health.decorators import register_healthcheck
+from health.healthchecks.base import BaseHealthCheck
+
+
+@register_healthcheck
+class CacheHealthCheck(BaseHealthCheck):
+    name = "cache"
+
+    def check(self, **kwargs) -> dict:
+        try:
+            cache.set("healthcheck", "ok", timeout=1)
+            result = cache.get("healthcheck")
+            if result == "ok":
+                return {"status": "ok"}
+            return {"status": "error", "details": "Cache misconfiguration."}
+        except Exception as e:
+            return {"status": "error", "details": str(e)}
+
+
+@register_healthcheck
+class DatabaseHealthCheck(BaseHealthCheck):
+    name = "database"
+
+    def check(self, **kwargs) -> dict:
+        try:
+            with connections["default"].cursor() as cursor:
+                cursor.execute("SELECT 1")
+            return {"status": "ok"}
+        except Exception as e:
+            return {"status": "error", "details": str(e)}
+
+
+@register_healthcheck
+class FileStorageHealthCheck(BaseHealthCheck):
+    name = "storage"
+
+    def check(self, **kwargs) -> dict:
+        try:
+            # Path for the healthcheck test file
+            test_file = "healthcheck/healthcheck_test_file.txt"
+            content = b"healthcheck"
+
+            # Ensure the "healthcheck" directory exists (for local file systems)
+            if not default_storage.exists("healthcheck/"):
+                default_storage.save(
+                    "healthcheck/.keep", ContentFile(b"")
+                )  # Create a placeholder file
+
+            # Write the test file
+            with default_storage.open(test_file, "wb") as f:
+                f.write(content)
+
+            # Verify the file exists
+            if not default_storage.exists(test_file):
+                return {"status": "error", "details": "File not found after creation."}
+
+            # Clean up: Delete the test file
+            default_storage.delete(test_file)
+
+            return {"status": "ok"}
+        except Exception as e:
+            return {"status": "error", "details": str(e)}
+
+
+@register_healthcheck
+class APIHealthCheck(BaseHealthCheck):
+    name = "api"
+
+    def check(self, **kwargs) -> dict:
+        request = kwargs.get("request")
+        try:
+            full_url = request.build_absolute_uri(reverse("api:ping-list"))
+            response = requests.get(full_url)
+            response.raise_for_status()
+            return {"status": "ok"}
+        except requests.RequestException as e:
+            return {"status": "error", "details": str(e)}

health/models.py

@@ -0,0 +1,3 @@
+from django.db import models
+
+# Create your models here.

health/registry.py

@@ -0,0 +1,29 @@
+from typing import List, Type
+
+from health.healthchecks.base import BaseHealthCheck
+
+
+class HealthCheckRegistry:
+    """
+    Registry to manage and store health checks for the entire project.
+    """
+
+    _registry: List[Type[BaseHealthCheck]] = []
+
+    @classmethod
+    def register(cls, healthcheck_class: Type[BaseHealthCheck]):
+        """
+        Register a health check class.
+        """
+        if not issubclass(healthcheck_class, BaseHealthCheck):
+            raise ValueError(
+                f"{healthcheck_class.__name__} must inherit from BaseHealthCheck."
+            )
+        cls._registry.append(healthcheck_class)
+
+    @classmethod
+    def get_registered_healthchecks(cls) -> List[BaseHealthCheck]:
+        """
+        Get a list of instantiated health checks from the registry.
+        """
+        return [healthcheck() for healthcheck in cls._registry]

health/templates/health/dashboard.html

@@ -0,0 +1,66 @@
+<!DOCTYPE html>
+<html lang="en">
+    <head>
+        <meta charset="UTF-8">
+        <meta name="viewport" content="width=device-width, initial-scale=1.0">
+        <title>Health | Data410</title>
+        <style>
+            body {
+                font-family: Arial, sans-serif;
+                margin: 2em;
+            }
+
+            main {
+                max-width: 800px;
+                margin: 0 auto;
+            }
+
+            h1 {
+                color: #2c3e50;
+                text-align: center;
+            }
+
+            p.timestamp {
+                text-align: center;
+                font-size: 0.9em;
+                color: #555;
+                margin-bottom: 1.5em;
+            }
+
+            ul {
+                list-style: none;
+                padding: 0;
+            }
+
+            li {
+                margin-bottom: 1em;
+                padding: 1em;
+                border: 1px solid #ddd;
+                border-radius: 5px;
+            }
+
+            .status-ok {
+                background-color: #dff0d8;
+                color: #3c763d;
+            }
+
+            .status-error {
+                background-color: #f2dede;
+                color: #a94442;
+            }
+        </style>
+    </head>
+    <body>
+        <main>
+            <h1>Data410 Health</h1>
+            <p class="timestamp">Last updated: {{ last_update }}</p>
+            <ul>
+                {% for name, result in results.items %}
+                    <li class="status-{{ result.status }}">
+                        <strong>{{ name }}</strong>: {{ result.status | upper }}
+                    </li>
+                {% endfor %}
+            </ul>
+        </main>
+    </body>
+</html>

health/tests.py

@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.

health/urls.py

@@ -0,0 +1,7 @@
+from django.urls import path, include
+
+from health.views import HealthCheckDashboardView
+
+urlpatterns = [
+    path("", HealthCheckDashboardView.as_view(), name="health-dashboard"),
+]

health/views.py

@@ -0,0 +1,74 @@
+from django.utils.timezone import now
+from django.conf import settings
+from django.core.cache import cache, caches
+from django.core.cache.backends.base import InvalidCacheBackendError
+from django.views.generic import TemplateView
+from health.registry import HealthCheckRegistry
+
+
+class HealthCheckDashboardView(TemplateView):
+    """
+    TemplateView to render the health check results in a user-friendly HTML template.
+    """
+
+    template_name = "health/dashboard.html"
+    CACHE_KEY = "healthcheck_dashboard"
+    CACHE_TIMEOUT = 30 * 60  # 30 minutes
+
+    def is_cache_available(self) -> bool:
+        """
+        Check if the cache backend is available and configured.
+        """
+        try:
+            caches["default"].get(self.CACHE_KEY)
+            return True
+        except:
+            return False
+
+    def generate_healthcheck_results(self, request) -> dict:
+        """
+        Generate fresh health check results.
+        """
+        healthchecks = HealthCheckRegistry.get_registered_healthchecks()
+        results = {
+            check.verbose_name: check.check(request=request) for check in healthchecks
+        }
+        last_update = now()
+        return {"results": results, "last_update": last_update}
+
+    def get_cached_healthcheck_results(self) -> dict:
+        """
+        Retrieve health check results from the cache.
+        """
+        cached_data = cache.get(self.CACHE_KEY)
+        return cached_data if cached_data else None
+
+    def cache_healthcheck_results(self, data: dict) -> None:
+        """
+        Cache health check results.
+        """
+        cache.set(self.CACHE_KEY, data, self.CACHE_TIMEOUT)
+
+    def get_context_data(self, **kwargs):
+        """
+        Prepare the context data for rendering the dashboard.
+        """
+        context = super().get_context_data(**kwargs)
+        request = self.request
+
+        if settings.DEBUG or not self.is_cache_available():
+            # Generate and use live health check results
+            data = self.generate_healthcheck_results(request)
+        else:
+            # Try to get cached results, fallback to generating fresh results
+            cached_data = self.get_cached_healthcheck_results()
+            if cached_data:
+                data = cached_data
+            else:
+                data = self.generate_healthcheck_results(request)
+                self.cache_healthcheck_results(data)
+
+        # Format the last update timestamp and add results to the context
+        context["results"] = data["results"]
+        context["last_update"] = data["last_update"].strftime("%Y-%m-%d %H:%M:%S %Z")
+        return context