feat: add HealthCheckService

django_healthy/_compat.py

@@ -1,11 +1,26 @@
 import sys
 
+if sys.version_info >= (3, 9):
+    from collections.abc import Mapping, MutableMapping
+else:
+    from typing import Mapping, MutableMapping
+
+if sys.version_info >= (3, 10):
+    from typing import TypeAlias
+else:
+    from typing_extensions import TypeAlias
+
 if sys.version_info >= (3, 11):
-    from typing import Self
+    from typing import NotRequired, Self, TypedDict
 else:
-    from typing_extensions import Self
+    from typing_extensions import NotRequired, Self, TypedDict
 
 
 __all__ = [
+    "Mapping",
+    "MutableMapping",
+    "NotRequired",
     "Self",
+    "TypeAlias",
+    "TypedDict",
 ]

django_healthy/health_checks/db.py

@@ -27,7 +27,7 @@ async def check_health(self) -> HealthCheckResult:
         else:
             return HealthCheckResult.healthy()
 
-    def _perform_health_check(self):
+    def _perform_health_check(self) -> None:
         connection: DjangoDatabaseWrapper = cast(DjangoDatabaseWrapper, connections[self.alias])
 
         with connection.cursor() as cursor:

django_healthy/health_checks/handler.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from typing import Any, Iterator, cast
+
+from django.conf import settings
+from django.core.exceptions import ImproperlyConfigured
+from django.utils.module_loading import import_string
+
+from django_healthy._compat import Mapping, MutableMapping, NotRequired, TypeAlias, TypedDict
+
+from .base import HealthCheck
+
+
+class InvalidHealthCheckError(ImproperlyConfigured):
+    pass
+
+
+class BackendConfig(TypedDict):
+    BACKEND: str
+    OPTIONS: NotRequired[Mapping[str, Any]]
+
+
+HealthCheckConfig: TypeAlias = Mapping[str, BackendConfig]
+
+
+class HealthCheckHandler(Mapping[str, HealthCheck]):
+    __slots__: tuple[str, ...] = ("_backends", "_health_checks")
+
+    def __init__(self, backends: HealthCheckConfig | None = None):
+        self._backends = (
+            backends if backends is not None else cast(HealthCheckConfig, getattr(settings, "HEALTH_CHECKS", {}))
+        )
+        self._health_checks: MutableMapping[str, HealthCheck] = {}
+
+    def __getitem__(self, alias: str) -> HealthCheck:
+        try:
+            return self._health_checks[alias]
+        except KeyError:
+            try:
+                params = self._backends[alias]
+            except KeyError as exc:
+                msg = f"Could not find config for '{alias}' in settings.HEALTH_CHECKS."
+                raise InvalidHealthCheckError(msg) from exc
+            else:
+                health_check = self.create_health_check(params)
+                self._health_checks[alias] = health_check
+                return health_check
+
+    def __iter__(self) -> Iterator[str]:
+        return iter(self._backends)
+
+    def __len__(self) -> int:
+        return len(self._backends)
+
+    def create_health_check(self, params: BackendConfig) -> HealthCheck:
+        backend = params["BACKEND"]
+        options = params.get("OPTIONS", {})
+
+        try:
+            factory = import_string(backend)
+        except ImportError as e:
+            msg = f"Could not find backend {backend!r}: {e}"
+            raise InvalidHealthCheckError(msg) from e
+        else:
+            return factory(**options)
+
+
+health_checks = HealthCheckHandler()

django_healthy/health_checks/service.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass, field
+from datetime import timedelta
+from timeit import default_timer as timer
+from typing import Any
+
+from django_healthy._compat import Mapping  # noqa: TCH001
+
+from .base import HealthCheck, HealthStatus
+from .handler import HealthCheckHandler, health_checks
+
+
+@dataclass
+class HealthReportEntry:
+    status: HealthStatus
+    duration: timedelta
+    description: str | None = None
+    exception: Exception | None = None
+    data: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class HealthReport:
+    entries: Mapping[str, HealthReportEntry]
+    status: HealthStatus
+    total_duration: timedelta
+
+
+class HealthCheckService:
+    __slots__: tuple[str, ...] = ("_handler",)
+
+    def __init__(self, handler: HealthCheckHandler | None = None):
+        self._handler = handler or health_checks
+
+    async def check_health(self) -> HealthReport:
+        start_time: float = timer()
+        task_map: dict[str, asyncio.Task[HealthReportEntry]] = {
+            name: asyncio.create_task(self.run_health_check(health_check))
+            for name, health_check in self._handler.items()
+        }
+        await asyncio.gather(*task_map.values())
+        end_time: float = timer()
+
+        entries: dict[str, HealthReportEntry] = {name: task.result() for name, task in task_map.items()}
+        worst_case = min(entry.status.value for entry in entries.values())
+        return HealthReport(
+            entries=entries,
+            status=HealthStatus(worst_case),
+            total_duration=timedelta(seconds=end_time - start_time),
+        )
+
+    async def run_health_check(self, health_check: HealthCheck) -> HealthReportEntry:
+        start_time: float = timer()
+        result = await health_check.check_health()
+        end_time: float = timer()
+
+        return HealthReportEntry(
+            status=result.status,
+            duration=timedelta(seconds=end_time - start_time),
+            description=result.description,
+            exception=result.exception,
+            data=result.data,
+        )

pyproject.toml

@@ -30,7 +30,7 @@ classifiers = [
 dependencies = [
   "asgiref>=3.3",
   "django>=4.2",
-  "typing-extensions; python_version<'3.11'"
+  "typing-extensions; python_version<'3.10'"
 ]
 
 [project.urls]

testproj/settings.py

@@ -100,3 +100,12 @@
 STATIC_URL = "static/"
 
 DEFAULT_AUTO_FIELD = "django.db.models.AutoField"
+
+HEALTH_CHECKS = {
+    "cache": {
+        "BACKEND": "django_healthy.health_checks.cache.CacheHealthCheck",
+    },
+    "db": {
+        "BACKEND": "django_healthy.health_checks.db.DatabasePingHealthCheck",
+    },
+}

tests/health_checks/test_handler.py

@@ -0,0 +1,50 @@
+import pytest
+from django.conf import settings
+
+from django_healthy.health_checks.cache import CacheHealthCheck
+from django_healthy.health_checks.handler import HealthCheckHandler, InvalidHealthCheckError
+
+
+class TestHealthCheckHandler:
+    def test_with_custom_settings(self):
+        handler = HealthCheckHandler(
+            backends={
+                "test": {
+                    "BACKEND": "django_healthy.health_checks.cache.CacheHealthCheck",
+                }
+            }
+        )
+        items = set(handler.keys())
+
+        assert items == {"test"}
+
+    def test_with_default_settings(self):
+        handler = HealthCheckHandler()
+        items = set(handler.keys())
+
+        assert items == set(settings.HEALTH_CHECKS)
+
+    def test_get_existing_item(self):
+        handler = HealthCheckHandler(
+            backends={
+                "test": {
+                    "BACKEND": "django_healthy.health_checks.cache.CacheHealthCheck",
+                }
+            }
+        )
+
+        got = handler["test"]
+
+        assert isinstance(got, CacheHealthCheck)
+
+    def test_get_missing_item(self):
+        handler = HealthCheckHandler(
+            backends={
+                "test": {
+                    "BACKEND": "django_healthy.health_checks.cache.CacheHealthCheck",
+                }
+            }
+        )
+
+        with pytest.raises(InvalidHealthCheckError):
+            handler["missing"]

tests/health_checks/test_service.py

@@ -0,0 +1,79 @@
+import pytest
+from django.conf import settings
+
+from django_healthy.health_checks import HealthStatus
+from django_healthy.health_checks.handler import HealthCheckHandler
+from django_healthy.health_checks.service import HealthCheckService
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.mark.asyncio
+class TestHealthCheckService:
+    async def test_with_default_handler(self):
+        service = HealthCheckService()
+
+        got = await service.check_health()
+
+        assert got.status == HealthStatus.HEALTHY
+        assert set(got.entries.keys()) == set(settings.HEALTH_CHECKS.keys())
+
+    async def test_with_custom_handler(self):
+        service = HealthCheckService(
+            HealthCheckHandler(
+                backends={
+                    "test": {
+                        "BACKEND": "django_healthy.health_checks.cache.CacheHealthCheck",
+                    }
+                }
+            )
+        )
+
+        got = await service.check_health()
+
+        assert got.status == HealthStatus.HEALTHY
+        assert set(got.entries.keys()) == {"test"}
+
+    async def test_with_unhealthy_service(self):
+        service = HealthCheckService(
+            HealthCheckHandler(
+                backends={
+                    "test": {
+                        "BACKEND": "django_healthy.health_checks.db.DatabasePingHealthCheck",
+                        "OPTIONS": {
+                            "alias": "dummy",
+                        },
+                    }
+                }
+            )
+        )
+
+        got = await service.check_health()
+
+        assert got.status == HealthStatus.UNHEALTHY
+        assert set(got.entries.keys()) == {"test"}
+
+    async def test_with_multiple_service_status_gets_worst_case(self):
+        service = HealthCheckService(
+            HealthCheckHandler(
+                backends={
+                    "healthy": {
+                        "BACKEND": "django_healthy.health_checks.db.DatabasePingHealthCheck",
+                        "OPTIONS": {
+                            "alias": "default",
+                        },
+                    },
+                    "unhealthy": {
+                        "BACKEND": "django_healthy.health_checks.db.DatabasePingHealthCheck",
+                        "OPTIONS": {
+                            "alias": "dummy",
+                        },
+                    },
+                }
+            )
+        )
+
+        got = await service.check_health()
+
+        assert got.status == HealthStatus.UNHEALTHY
+        assert set(got.entries.keys()) == {"healthy", "unhealthy"}