feat: add more light weight database check (#439)

* feat: add db_heartbeat

* feat: add tests

* feat: add ignore .DS_Store

* fix: linting errors

* fix: linting errors

* fix: linting errors

* feat: remove .ds_store

* feat: add tests

* fix: test linting

* feat: fix incorrect imports

* feat: add test mocking

* fix: linting

* fix: linting

* fix: tests

* fix: linting

* feat: update docs

Author: panteparak

.gitignore

@@ -105,4 +105,6 @@ ENV/
 
 .envrc
 .direnv
+
+# mac
 .DS_Store

README.md

@@ -20,6 +20,7 @@ The following health checks are bundled with this project:
 - Celery ping
 - RabbitMQ
 - Migrations
+- Database Heartbeat (Lightweight version of `health_check.db`)
 
 Writing your own custom health checks is also very quick and easy.
 
@@ -74,6 +75,7 @@ Add the `health_check` applications to your `INSTALLED_APPS`:
         'health_check.contrib.s3boto3_storage',     # requires boto3 and S3BotoStorage backend
         'health_check.contrib.rabbitmq',            # requires RabbitMQ broker
         'health_check.contrib.redis',               # requires Redis broker
+        'health_check.contrib.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,12 @@
+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,22 @@
+from django.db import connection
+
+from health_check.backends import BaseHealthCheckBackend
+from health_check.exceptions import ServiceUnavailable
+
+
+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 ServiceUnavailable(
+                    "Health Check query did not return the expected result."
+                )
+        except Exception as e:
+            raise ServiceUnavailable(f"Database health check failed: {e}")

tests/test_db_heartbeat.py

@@ -0,0 +1,27 @@
+import unittest
+from unittest.mock import MagicMock, patch
+
+from health_check.contrib.db_heartbeat.backends import DatabaseHeartBeatCheck
+from health_check.exceptions import ServiceUnavailable
+
+
+class TestDatabaseHeartBeatCheck(unittest.TestCase):
+    @patch("health_check.contrib.db_heartbeat.backends.connection")
+    def test_check_status_success(self, mock_connection):
+        mock_cursor = MagicMock()
+        mock_cursor.fetchone.return_value = (1,)
+        mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
+
+        health_check = DatabaseHeartBeatCheck()
+        try:
+            health_check.check_status()
+        except Exception as e:
+            self.fail(f"check_status() raised an exception unexpectedly: {e}")
+
+    @patch("health_check.contrib.db_heartbeat.backends.connection")
+    def test_check_status_service_unavailable(self, mock_connection):
+        mock_connection.cursor.side_effect = Exception("Database error")
+
+        health_check = DatabaseHeartBeatCheck()
+        with self.assertRaises(ServiceUnavailable):
+            health_check.check_status()

tests/testapp/settings.py

@@ -30,6 +30,7 @@
     "health_check.contrib.migrations",
     "health_check.contrib.celery_ping",
     "health_check.contrib.s3boto_storage",
+    "health_check.contrib.db_heartbeat",
     "tests",
 )