diff --git a/django_healthy/_compat.py b/django_healthy/_compat.py new file mode 100644 index 0000000..ab5b393 --- /dev/null +++ b/django_healthy/_compat.py @@ -0,0 +1,11 @@ +import sys + +if sys.version_info >= (3, 11): + from typing import Self +else: + from typing_extensions import Self + + +__all__ = [ + "Self", +] diff --git a/django_healthy/health_checks/__init__.py b/django_healthy/health_checks/__init__.py new file mode 100644 index 0000000..e6e7b26 --- /dev/null +++ b/django_healthy/health_checks/__init__.py @@ -0,0 +1,7 @@ +from .base import HealthCheck, HealthCheckResult, HealthStatus + +__all__ = [ + "HealthCheck", + "HealthCheckResult", + "HealthStatus", +] diff --git a/django_healthy/health_checks/base.py b/django_healthy/health_checks/base.py new file mode 100644 index 0000000..fd247c7 --- /dev/null +++ b/django_healthy/health_checks/base.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from enum import Enum, auto +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from django_healthy._compat import Self + + +class HealthStatus(Enum): + UNHEALTHY = auto() + DEGRADED = auto() + HEALTHY = auto() + + +@dataclass +class HealthCheckResult: + status: HealthStatus + description: str | None = None + exception: Exception | None = None + data: dict[str, Any] = field(default_factory=dict) + + @classmethod + def healthy( + cls, + description: str | None = None, + exception: Exception | None = None, + data: dict[str, Any] | None = None, + ) -> Self: + return cls( + status=HealthStatus.HEALTHY, + description=description, + exception=exception, + data=data or {}, + ) + + @classmethod + def degraded( + cls, + description: str | None = None, + exception: Exception | None = None, + data: dict[str, Any] | None = None, + ) -> Self: + return cls( + status=HealthStatus.DEGRADED, + description=description, + exception=exception, + data=data or {}, + ) + + @classmethod + def unhealthy( + cls, + description: str | None = None, + exception: Exception | None = None, + data: dict[str, Any] | None = None, + ) -> Self: + return cls( + status=HealthStatus.UNHEALTHY, + description=description, + exception=exception, + data=data or {}, + ) + + +class HealthCheck(ABC): + @abstractmethod + async def check_health(self) -> HealthCheckResult: ... diff --git a/pyproject.toml b/pyproject.toml index 1fcedee..47b571d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,6 +29,7 @@ classifiers = [ ] dependencies = [ "django>=4.2", + "typing-extensions; python_version<'3.11'" ] [project.urls] @@ -68,6 +69,7 @@ source = "vcs" [tool.hatch.envs.default] dependencies = [ "coverage[toml]>=6.5", + "faker", "pytest", "pytest-asyncio", "pytest-django", diff --git a/tests/health_checks/__init__.py b/tests/health_checks/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/health_checks/test_base.py b/tests/health_checks/test_base.py new file mode 100644 index 0000000..9e82575 --- /dev/null +++ b/tests/health_checks/test_base.py @@ -0,0 +1,75 @@ +from django_healthy.health_checks.base import HealthCheckResult, HealthStatus + + +class TestHealthCheckResult: + def test_healthy_defaults(self): + got = HealthCheckResult.healthy() + + assert got.status == HealthStatus.HEALTHY + assert got.description is None + assert got.exception is None + assert got.data == {} + + def test_healthy_params(self, faker): + given_description = faker.sentence() + given_exception = Exception() + given_data = faker.pydict() + + got = HealthCheckResult.healthy( + description=given_description, + exception=given_exception, + data=given_data, + ) + + assert got.status == HealthStatus.HEALTHY + assert got.description == given_description + assert got.exception == given_exception + assert got.data == given_data + + def test_degraded_defaults(self): + got = HealthCheckResult.degraded() + + assert got.status == HealthStatus.DEGRADED + assert got.description is None + assert got.exception is None + assert got.data == {} + + def test_degraded_params(self, faker): + given_description = faker.sentence() + given_exception = Exception() + given_data = faker.pydict() + + got = HealthCheckResult.degraded( + description=given_description, + exception=given_exception, + data=given_data, + ) + + assert got.status == HealthStatus.DEGRADED + assert got.description == given_description + assert got.exception == given_exception + assert got.data == given_data + + def test_unhealthy_defaults(self): + got = HealthCheckResult.unhealthy() + + assert got.status == HealthStatus.UNHEALTHY + assert got.description is None + assert got.exception is None + assert got.data == {} + + def test_unhealthy_params(self, faker): + given_description = faker.sentence() + given_exception = Exception() + given_data = faker.pydict() + + got = HealthCheckResult.unhealthy( + description=given_description, + exception=given_exception, + data=given_data, + ) + + assert got.status == HealthStatus.UNHEALTHY + assert got.description == given_description + assert got.exception == given_exception + assert got.data == given_data