feat: add db_heartbeat

health_check/contrib/db_heartbeat/__init__.py

@@ -0,0 +1,4 @@
+import django
+
+if django.VERSION < (3, 2):
+    default_app_config = "health_check.contrib.db_heartbeat.apps.HealthCheckConfig"

health_check/contrib/db_heartbeat/apps.py

@@ -0,0 +1,10 @@
+from django.apps import AppConfig
+from health_check.plugins import plugin_dir
+
+
+class HealthCheckConfig(AppConfig):
+    name = 'health_check.contrib.db_heartbeat'
+
+    def ready(self):
+        from .backends import DatabaseHeartBeatCheck
+        plugin_dir.register(DatabaseHeartBeatCheck)

health_check/contrib/db_heartbeat/backends.py

@@ -0,0 +1,21 @@
+from health_check.backends import BaseHealthCheckBackend
+from health_check.exceptions import ServiceUnavailable, ServiceReturnedUnexpectedResult
+from django.db import connection
+
+
+class DatabaseHeartBeatCheck(BaseHealthCheckBackend):
+    """
+    Health check that runs a simple SELECT 1; query to test if the database connection is alive.
+    """
+
+    def check_status(self):
+        try:
+            result = None
+            with connection.cursor() as cursor:
+                cursor.execute("SELECT 1;")
+                result = cursor.fetchone()
+
+            if result != (1,):
+                raise ServiceReturnedUnexpectedResult("Health Check query did not return the expected result.")
+        except Exception as e:
+            raise ServiceUnavailable(f"Database health check failed: {e}")