fix: migrate custom_health_checks to django-health-check v4 API

Replace BaseHealthCheckBackend/plugin_dir/MainView with HealthCheck/
HealthCheckView. Switch check_status() to run() and register checks
via HealthCheckView.checks instead of plugin_dir.

Refs: NS-209

Author: mhieta

custom_health_checks/apps.py

@@ -1,16 +1,5 @@
-import logging
-
 from django.apps import AppConfig
-from health_check.plugins import plugin_dir
-
-logger = logging.getLogger(__name__)
 
 
 class CustomHealthChecksAppConfig(AppConfig):
     name = "custom_health_checks"
-
-    def ready(self):
-        from .backends import DatabaseHealthCheck
-
-        plugin_dir.register(DatabaseHealthCheck)
-        logger.info("Registered DatabaseHealthCheck to health_check plugins.")

custom_health_checks/backends.py

@@ -1,16 +1,17 @@
+import dataclasses
+
 from django.db import connection
-from health_check.backends import BaseHealthCheckBackend
+from health_check.base import HealthCheck
 from health_check.exceptions import ServiceUnavailable
 
 
-class DatabaseHealthCheck(BaseHealthCheckBackend):
+@dataclasses.dataclass
+class DatabaseHealthCheck(HealthCheck):
     """
     Custom health check for the database connection.
     """
 
-    critical_service = True
-
-    def check_status(self):
+    def run(self):
         """
         Checks the database connection by executing a simple query.
         """
@@ -19,6 +20,3 @@ def check_status(self):
                 cursor.execute("SELECT 1 FROM users_user LIMIT 1")
         except Exception as e:
             raise ServiceUnavailable(f"Database connection failed: {e}")
-
-    def identifier(self):
-        return self.__class__.__name__  # Display name on the endpoint.

custom_health_checks/tests/test_backends.py

@@ -10,35 +10,34 @@
 @patch("django.db.connection.cursor")
 def test_database_check_status_success(mock_cursor):
     """
-    Test that check_status passes when the database connection is successful.
+    Test that run passes when the database connection is successful.
     """
     mock_cursor.return_value.__enter__.return_value.execute.return_value = True
     health_check = DatabaseHealthCheck()
-    health_check.check_status()  # Should not raise an exception
+    health_check.run()  # Should not raise an exception
 
 
 @patch("django.db.connection.cursor")
 def test_database_check_status_failure(mock_cursor):
     """
-    Test that check_status raises ServiceUnavailable when the database
-    connection fails.
+    Test that run raises ServiceUnavailable when the database connection fails.
     """
     mock_cursor.return_value.__enter__.return_value.execute.side_effect = (
         OperationalError("Database error")
     )
     health_check = DatabaseHealthCheck()
     with pytest.raises(ServiceUnavailable) as exc_info:
-        health_check.check_status()
+        health_check.run()
     assert "Database connection failed:" in str(exc_info.value)
 
 
 def test_database_check_status_with_real_database(db):  # Use the 'db' fixture
     """
-    Test check_status with the actual database connection.
+    Test run with the actual database connection.
     This assumes your test database is set up correctly.
     """
     try:
         health_check = DatabaseHealthCheck()
-        health_check.check_status()  # Should not raise an exception
+        health_check.run()  # Should not raise an exception
     except ServiceUnavailable as e:
         pytest.fail(f"Database health check failed: {e}")

custom_health_checks/tests/test_healthz.py

@@ -5,24 +5,24 @@
 from health_check.exceptions import ServiceUnavailable
 
 
-@patch("custom_health_checks.backends.DatabaseHealthCheck.check_status")
-def test_healthz_success(mock_check_status, client: Client):  # Use the 'client' fixture
+@patch("custom_health_checks.backends.DatabaseHealthCheck.run")
+def test_healthz_success(mock_run, client: Client):  # Use the 'client' fixture
     """
     Test /healthz endpoint with successful health checks.
     """
-    mock_check_status.return_value = None  # Simulate successful check
+    mock_run.return_value = None  # Simulate successful check
     url = reverse("healthz")
     response = client.get(url)
     assert response.status_code == 200
-    assert response.content == b'{"DatabaseHealthCheck": "working"}'
+    assert response.json() == {"DatabaseHealthCheck()": "OK"}
 
 
-@patch("custom_health_checks.backends.DatabaseHealthCheck.check_status")
-def test_healthz_database_error(mock_check_status, client: Client):
+@patch("custom_health_checks.backends.DatabaseHealthCheck.run")
+def test_healthz_database_error(mock_run, client: Client):
     """
     Test /healthz endpoint with a database error.
     """
-    mock_check_status.side_effect = ServiceUnavailable("Database error")
+    mock_run.side_effect = ServiceUnavailable("Database error")
     url = reverse("healthz")
     response = client.get(url)
     assert response.status_code == 500

custom_health_checks/views.py

@@ -1,13 +1,13 @@
+import asyncio
+
 from django.utils.decorators import method_decorator
 from django.views.decorators.cache import never_cache
-from health_check.views import MainView
+from health_check.views import HealthCheckView
 
 
-class HealthCheckJSONView(MainView):
+class HealthCheckJSONView(HealthCheckView):
     """
-    The original `health_check` view uses format-attribute or headers to determine
-    what response type is used. The HealthCheckJSONView can be used to ensure
-    that a JSON Response is always used (for health check reporting).
+    Always returns a JSON response, regardless of the Accept header.
 
     To apply it, in project's `urls.py`, add the custom JSON view in use like this:
     >>> # doctest: +SKIP
@@ -17,17 +17,20 @@ class HealthCheckJSONView(MainView):
     ...     # ...
     ...     path(
     ...         r"healthz",
-    ...         views.HealthCheckCustomView.as_view(),
+    ...         views.HealthCheckJSONView.as_view(),
     ...         name="health_check_custom",
     ...     ),
     ... ]
     """
 
+    checks = ("custom_health_checks.backends.DatabaseHealthCheck",)
+
     @method_decorator(never_cache)
-    def get(self, request, *args, **kwargs):
-        subset = kwargs.get("subset", None)
-        health_check_has_error = self.check(subset)
-        status_code = 500 if health_check_has_error else 200
-        return self.render_to_response_json(
-            self.filter_plugins(subset=subset), status_code
-        )
+    async def get(self, request, *args, **kwargs):
+        with self.get_executor() as executor:
+            self.results = await asyncio.gather(
+                *(check.get_result(executor) for check in self.get_checks())
+            )
+        has_errors = any(result.error for result in self.results)
+        status_code = 500 if has_errors else 200
+        return self.render_to_response_json(status_code)