Checks

Django Built-in Services

health_check.Cache dataclass

Bases: HealthCheck

Check that the cache backend is able to set and get a value.

It can be setup multiple times for different cache aliases if needed.

Parameters:

Name	Type	Description	Default
alias	str	The cache alias to test against.	'default'
key_prefix	str	Prefix for the node specific cache key.	'djangohealthcheck_test'
timeout	timedelta	Time until probe keys expire in the cache backend.	timedelta(seconds=5)

Source code in health_check/checks.py

@dataclasses.dataclass
class Cache(HealthCheck):
    """
    Check that the cache backend is able to set and get a value.

    It can be setup multiple times for different cache aliases if needed.

    Args:
        alias: The cache alias to test against.
        key_prefix: Prefix for the node specific cache key.
        timeout: Time until probe keys expire in the cache backend.

    """

    alias: str = "default"
    key_prefix: str = dataclasses.field(default="djangohealthcheck_test", repr=False)
    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=5), repr=False
    )

    async def run(self):
        try:
            cache = caches[self.alias]
        except InvalidCacheBackendError as e:
            msg = "Cache alias does not exist"
            raise ServiceUnavailable(msg) from e
        # Use an isolated key per probe run to avoid cross-process write races.
        cache_key = f"{self.key_prefix}:{uuid.uuid4().hex}"
        cache_value = f"itworks-{datetime.datetime.now(tz=datetime.UTC).timestamp()}"
        try:
            await cache.aset(
                cache_key,
                cache_value,
                timeout=self.timeout.total_seconds(),
            )
            if await cache.aget(cache_key) != cache_value:
                msg = f"Cache key {cache_key} does not match"
                raise ServiceUnavailable(msg)
        except CacheKeyWarning as e:
            msg = "Cache key warning"
            raise ServiceReturnedUnexpectedResult(msg) from e
        except ValueError as e:
            raise ServiceReturnedUnexpectedResult("ValueError") from e
        except (ConnectionError, RedisError) as e:
            msg = "Connection Error"
            raise ServiceReturnedUnexpectedResult(msg) from e

health_check.DNS dataclass

Bases: HealthCheck

Check DNS resolution by resolving the server's hostname.

Verifies that DNS resolution is working using the system's configured DNS servers, as well as nameserver resolution for the provided hostname.

Parameters:

Name	Type	Description	Default
hostname	str	The hostname to resolve.	gethostname()
timeout	timedelta	DNS query timeout.	timedelta(seconds=5)
nameservers	Sequence[str | Nameserver]	Optional list of nameservers to use for DNS resolution.	(lambda: nameservers)()
record_type	str | RdataType	DNS record type to query (default is A record).	A

Source code in health_check/checks.py

@dataclasses.dataclass
class DNS(HealthCheck):
    """
    Check DNS resolution by resolving the server's hostname.

    Verifies that DNS resolution is working using the system's configured
    DNS servers, as well as nameserver resolution for the provided hostname.

    Args:
        hostname: The hostname to resolve.
        timeout: DNS query timeout.
        nameservers: Optional list of nameservers to use for DNS resolution.
        record_type: DNS record type to query (default is A record).

    """

    hostname: str = dataclasses.field(default_factory=socket.gethostname)
    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=5), repr=False
    )
    nameservers: typing.Sequence[str | Nameserver] = dataclasses.field(
        default_factory=lambda: dns.resolver.get_default_resolver().nameservers
    )
    record_type: str | dns.rdatatype.RdataType = dataclasses.field(
        default=dns.rdatatype.A
    )

    async def run(self):
        logger.debug("Attempting to resolve hostname: %s", self.hostname)

        resolver = dns.asyncresolver.Resolver()
        resolver.lifetime = self.timeout.total_seconds()
        resolver.nameservers = [*self.nameservers]

        try:
            # Perform DNS resolution (A record by default)
            answers = await resolver.resolve(self.hostname, self.record_type)
        except dns.resolver.NXDOMAIN as e:
            msg = f"DNS resolution failed: hostname {self.hostname} does not exist"
            raise ServiceUnavailable(msg) from e
        except dns.resolver.NoAnswer as e:
            msg = f"DNS resolution failed: no answer for {self.hostname}"
            raise ServiceUnavailable(msg) from e
        except dns.resolver.Timeout as e:
            msg = f"DNS resolution failed: timeout resolving {self.hostname}"
            raise ServiceUnavailable(msg) from e
        except dns.resolver.NoNameservers as e:
            msg = "DNS resolution failed: no nameservers available"
            raise ServiceUnavailable(msg) from e
        except dns.exception.DNSException as e:
            msg = f"DNS resolution failed: {e}"
            raise ServiceUnavailable(msg) from e
        else:
            logger.debug(
                "Successfully resolved %s to %s",
                self.hostname,
                [str(rdata) for rdata in answers],
            )

health_check.Database dataclass

Bases: HealthCheck

Check database operation by executing a simple SELECT 1 query.

It can be setup multiple times for different database connections if needed. No actual data is read from or written to the database to minimize the performance impact and work with conservative database user permissions.

Parameters:

Name	Type	Description	Default
alias	str	The alias of the database connection to check.	'default'

Source code in health_check/checks.py

@dataclasses.dataclass
class Database(HealthCheck):
    """
    Check database operation by executing a simple SELECT 1 query.

    It can be setup multiple times for different database connections if needed.
    No actual data is read from or written to the database to minimize the performance impact
    and work with conservative database user permissions.

    Args:
        alias: The alias of the database connection to check.

    """

    alias: str = "default"

    def run(self):
        try:
            connection = connections[self.alias]
        except ConnectionDoesNotExist as e:
            msg = "Database alias does not exist"
            raise ServiceUnavailable(msg) from e
        try:
            compiler = connection.ops.compiler("SQLCompiler")(
                _SelectOne(), connection, None
            )
            with connection.temporary_connection() as cursor:
                cursor.execute(*compiler.compile(_SelectOne()))
                result = cursor.fetchone()
        except db.Error as e:
            raise ServiceUnavailable(str(e).rsplit(":")[0]) from e
        else:
            if result != (1,):
                msg = "Health Check query did not return the expected result."
                raise ServiceUnavailable(msg)

health_check.Mail dataclass

Bases: HealthCheck

Check that an email backend is able to open and close the connection.

Parameters:

Name	Type	Description	Default
backend	str | None	The email backend to test against. Legacy, used only on Django < 6.1.	None
alias	str	The mailer alias to test.	DEFAULT_MAILER_ALIAS
timeout	timedelta	Timeout for connection to an email server in seconds.	timedelta(seconds=15)

Source code in health_check/checks.py

@dataclasses.dataclass
class Mail(HealthCheck):
    """
    Check that an email backend is able to open and close the connection.

    Args:
        backend: The email backend to test against. Legacy, used only on Django < 6.1.
        alias: The mailer alias to test.
        timeout: Timeout for connection to an email server in seconds.

    """

    backend: str | None = dataclasses.field(default=None, repr=False)
    alias: str = DEFAULT_MAILER_ALIAS
    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=15), repr=False
    )

    def _get_connection(self) -> BaseEmailBackend:
        if django.VERSION >= (6, 1):
            return mailers[self.alias]
        return get_connection(self.backend, fail_silently=False)

    def run(self) -> None:
        connection: BaseEmailBackend = self._get_connection()
        connection.timeout = self.timeout.total_seconds()
        logger.debug("Trying to open connection to mail backend.")
        try:
            connection.open()
        except smtplib.SMTPException as e:
            msg = "Failed to open connection with SMTP server"
            raise ServiceUnavailable(msg) from e
        except ConnectionRefusedError as e:
            msg = "Connection refused error"
            raise ServiceUnavailable(msg) from e
        finally:
            connection.close()
        logger.debug("Connection established. Mail backend %r is healthy.", connection)

health_check.RemoteCheck dataclass

Bases: HealthCheck

Check a remote django-health-check endpoint.

Report every non-2xx response as a single error.

Parameters:

Name	Type	Description	Default
url	str	Absolute URL of the remote health check.	required
auth_token	str | None	Bearer token for an authenticated remote health check.	None
label	str | None	Label that replaces the URL in the check representation.	None
timeout	timedelta	Request timeout.	timedelta(seconds=10)

Source code in health_check/checks.py

@dataclasses.dataclass
class RemoteCheck(HealthCheck):
    """
    Check a remote django-health-check endpoint.

    Report every non-2xx response as a single error.

    Args:
        url: Absolute URL of the remote health check.
        auth_token: Bearer token for an authenticated remote health check.
        label: Label that replaces the URL in the check representation.
        timeout: Request timeout.

    """

    url: str = dataclasses.field(repr=False)
    auth_token: str | None = dataclasses.field(default=None, repr=False)
    label: str | None = None
    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )

    def __repr__(self) -> str:
        match self.label:
            case None:
                return f"{type(self).__name__}(url={self.url!r})"
            case label:
                return f"{type(self).__name__}(label={label!r})"

    async def run(self) -> None:
        headers = {
            "Accept": "application/json",
            "User-Agent": "django-health-check",
        }
        if token := self.auth_token:
            headers["Authorization"] = f"Bearer {token}"

        async with httpx.AsyncClient(
            headers=headers,
            timeout=self.timeout.total_seconds(),
            follow_redirects=True,
        ) as client:
            try:
                response = await client.get(self.url)
            except httpx.TimeoutException as e:
                msg = "Request timed out"
                raise ServiceUnavailable(msg) from e
            except httpx.RequestError as e:
                msg = f"Failed to reach URL: {e}"
                raise ServiceUnavailable(msg) from e

        match response.status_code // 100:
            case 2:
                return
            case 5:
                raise RemoteServiceUnavailableError(response)
            case _:
                raise RemoteServiceReturnedUnexpectedResultError(response.status_code)

health_check.Storage dataclass

Bases: HealthCheck

Check file storage backends by saving, reading, and deleting a test file.

It can be set up multiple times for different storage backends if needed.

Parameters:

Name	Type	Description	Default
alias	str	The alias of the storage backend to check.	'default'

Source code in health_check/checks.py

@dataclasses.dataclass
class Storage(HealthCheck):
    """
    Check file storage backends by saving, reading, and deleting a test file.

    It can be set up multiple times for different storage backends if needed.

    Args:
        alias: The alias of the storage backend to check.

    """

    alias: str = "default"

    @property
    def storage(self) -> DjangoStorage:
        try:
            return storages[self.alias]
        except InvalidStorageError as e:
            msg = "Storage alias does not exist"
            raise ServiceUnavailable(msg) from e

    def get_file_name(self):
        return f"health_check_storage_test/test-{uuid.uuid4()}.txt"

    def get_file_content(self):
        return f"# generated by health_check.Storage at {datetime.datetime.now(tz=datetime.UTC).timestamp()}".encode()

    def check_save(self, file_name, file_content):
        # save the file
        file_name = self.storage.save(file_name, ContentFile(content=file_content))
        # read the file and compare
        if not self.storage.exists(file_name):
            msg = "File does not exist"
            raise ServiceUnavailable(msg)
        with self.storage.open(file_name) as f:
            if f.read() != file_content:
                msg = "File content does not match"
                raise ServiceUnavailable(msg)
        return file_name

    def check_delete(self, file_name):
        # delete the file and make sure it is gone
        self.storage.delete(file_name)
        if self.storage.exists(file_name):
            msg = "File was not deleted"
            raise ServiceUnavailable(msg)

    def run(self):
        # write the file to the storage backend
        file_name = self.get_file_name()
        file_content = self.get_file_content()
        try:
            file_name = self.check_save(file_name, file_content)
            self.check_delete(file_name)
        finally:
            self.storage.delete(file_name)

System Services

To use the psutil-based checks, install the psutil extra:

pip install django-health-check[psutil]

health_check.contrib.psutil.Battery dataclass

Bases: HealthCheck

Warn about system battery status and power connection.

Parameters:

Name	Type	Description	Default
min_percent_available	float | None	Minimum battery percentage available or None to disable the check.	20.0
power_plugged	bool	Whether to warn if the power is unplugged.	False

Source code in health_check/contrib/psutil.py

@dataclasses.dataclass
class Battery(HealthCheck):
    """
    Warn about system battery status and power connection.

    Args:
        min_percent_available: Minimum battery percentage available or None to disable the check.
        power_plugged: Whether to warn if the power is unplugged.

    """

    min_percent_available: float | None = dataclasses.field(default=20.0, repr=False)
    power_plugged: bool = dataclasses.field(default=False, repr=False)
    hostname: str = dataclasses.field(default_factory=socket.gethostname, init=False)

    def run(self):
        try:
            battery = psutil.sensors_battery()
        except AttributeError as e:
            msg = "Battery information not available"
            raise ServiceUnavailable(msg) from e
        except ValueError as e:
            raise ServiceReturnedUnexpectedResult("ValueError") from e
        else:
            if (
                self.min_percent_available
                and battery.percent <= self.min_percent_available
            ):
                msg = f"Battery {battery.percent:.1f}\u202f%"
                raise ServiceWarning(msg)
            if self.power_plugged and not battery.power_plugged:
                msg = f"Power unplugged with battery at {battery.percent:.1f}\u202f%"
                raise ServiceWarning(msg)

health_check.contrib.psutil.CPU dataclass

Bases: HealthCheck

Warn about system CPU utilization.

The utilization represents an interval rather than a point in time measurement. The interval starts with the previous execution of this check and with the current unless an explicit interval is provided. An explicit interval will cause a blocking measurement and increases the execution time of the check by the length of the interval.

Parameters:

Name	Type	Description	Default
max_usage_percent	float | None	Maximum CPU usage in percent or None to disable the check.	80.0
interval	timedelta | None	The interval to measure CPU usage over or None to use the interval since the last check execution.	None

Source code in health_check/contrib/psutil.py

@dataclasses.dataclass
class CPU(HealthCheck):
    """
    Warn about system CPU utilization.

    The utilization represents an interval rather than a point in time measurement.
    The interval starts with the previous execution of this check and with the current
    unless an explicit interval is provided. An explicit interval will cause a blocking
    measurement and increases the execution time of the check by the length of the interval.

    Args:
        max_usage_percent: Maximum CPU usage in percent or None to disable the check.
        interval: The interval to measure CPU usage over or None to use the interval since the last check execution.

    """

    max_usage_percent: float | None = dataclasses.field(default=80.0, repr=False)
    interval: datetime.timedelta | None = dataclasses.field(default=None, repr=False)
    hostname: str = dataclasses.field(default_factory=socket.gethostname, init=False)

    def run(self):
        try:
            usage_percent = psutil.cpu_percent(
                interval=self.interval.total_seconds() if self.interval else None
            )
            msg = f"CPU {usage_percent:.1f}\u202f%"
            if self.max_usage_percent and usage_percent >= self.max_usage_percent:
                raise ServiceWarning(msg)
        except ValueError as e:
            raise ServiceReturnedUnexpectedResult("ValueError") from e

health_check.contrib.psutil.Memory dataclass

Bases: HealthCheck

Warn about system memory utilization.

Parameters:

Name	Type	Description	Default
min_gibibytes_available	float | None	Minimum available memory in gibibytes or None to disable the check.	None
max_memory_usage_percent	float | None	Maximum memory usage in percent or None to disable the check.	90.0

Source code in health_check/contrib/psutil.py

@dataclasses.dataclass()
class Memory(HealthCheck):
    """
    Warn about system memory utilization.

    Args:
        min_gibibytes_available: Minimum available memory in gibibytes or None to disable the check.
        max_memory_usage_percent: Maximum memory usage in percent or None to disable the check.

    """

    min_gibibytes_available: float | None = dataclasses.field(default=None, repr=False)
    max_memory_usage_percent: float | None = dataclasses.field(default=90.0, repr=False)
    hostname: str = dataclasses.field(default_factory=socket.gethostname, init=False)

    def run(self):
        try:
            try:
                memory = _CGroup()
            except OSError, ValueError:
                memory = psutil.virtual_memory()
            available_gibi = memory.available / (1024**3)
            total_gibi = memory.total / (1024**3)
            msg = f"RAM {available_gibi:.1f}/{total_gibi:.1f}GiB ({memory.percent:.1f}\u202f%)"
            if (
                self.min_gibibytes_available
                and available_gibi < self.min_gibibytes_available
            ):
                raise ServiceWarning(msg)
            if (
                self.max_memory_usage_percent
                and memory.percent >= self.max_memory_usage_percent
            ):
                raise ServiceWarning(msg)
        except ValueError as e:
            raise ServiceReturnedUnexpectedResult("ValueError") from e

health_check.contrib.psutil.Disk dataclass

Bases: HealthCheck

Warn about disk usage for a given system path.

It can be setup multiple times at different system paths, e.g. one at your application root and one at your media storage root.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to check disk usage for.	getcwd()
max_disk_usage_percent	float | None	Maximum disk usage in percent or None to disable the check.	90.0

Source code in health_check/contrib/psutil.py

@dataclasses.dataclass()
class Disk(HealthCheck):
    """
    Warn about disk usage for a given system path.

    It can be setup multiple times at different system paths,
    e.g. one at your application root and one at your media storage root.

    Args:
        path: Path to check disk usage for.
        max_disk_usage_percent: Maximum disk usage in percent or None to disable the check.

    """

    path: pathlib.Path | str = dataclasses.field(default_factory=os.getcwd)
    max_disk_usage_percent: float | None = dataclasses.field(default=90.0, repr=False)
    hostname: str = dataclasses.field(default_factory=socket.gethostname, init=False)

    def run(self):
        try:
            du = psutil.disk_usage(str(self.path))
            if (
                self.max_disk_usage_percent
                and du.percent >= self.max_disk_usage_percent
            ):
                msg = f"{du.percent}\u202f% disk usage"
                raise ServiceWarning(msg)
        except ValueError as e:
            raise ServiceReturnedUnexpectedResult("ValueError") from e

health_check.contrib.psutil.Temperature dataclass

Bases: HealthCheck

Warn about system temperature.

Each sensor's current temperature is compared to a maximum. Without a configured maximum, each sensor's high threshold applies. A sensor without a high threshold is skipped unless a maximum is configured. A configured maximum above a sensor's critical threshold is reported as a warning.

Parameters:

Name	Type	Description	Default
device	str | None	The device to check temperature for, e.g. 'coretemp' for CPU temperature on many systems. If None, check all available sensors.	'coretemp'
max_temperature_celsius	float | None | object	Maximum temperature in degree Celsius. If omitted, use each sensor's high threshold. Set to None to disable the check.	SENSOR_HIGH_THRESHOLD

Source code in health_check/contrib/psutil.py

@dataclasses.dataclass
class Temperature(HealthCheck):
    """
    Warn about system temperature.

    Each sensor's current temperature is compared to a maximum.
    Without a configured maximum, each sensor's high threshold applies.
    A sensor without a high threshold is skipped unless a maximum is configured.
    A configured maximum above a sensor's critical threshold is reported as a warning.

    Args:
        device: The device to check temperature for, e.g. 'coretemp' for CPU temperature on many systems. If None, check all available sensors.
        max_temperature_celsius: Maximum temperature in degree Celsius. If omitted, use each sensor's high threshold. Set to None to disable the check.

    """

    device: str | None = dataclasses.field(default="coretemp")
    max_temperature_celsius: float | None | object = dataclasses.field(
        default=SENSOR_HIGH_THRESHOLD, repr=False
    )
    hostname: str = dataclasses.field(default_factory=socket.gethostname, init=False)

    def run(self):
        try:
            temperatures = psutil.sensors_temperatures()
        except AttributeError as e:
            msg = "Temperature information not available"
            raise ServiceUnavailable(msg) from e
        if self.device:
            self._check_device(temperatures)
        else:
            for device, sensors in temperatures.items():
                for sensor in sensors:
                    self._check_temperature(sensor, device)

    def _check_device(self, temperatures):
        """Warn about every overheating sensor of the configured device."""
        try:
            sensors = temperatures[self.device]
        except KeyError:
            msg = f"Sensor {self.device!r} not found"
            raise ServiceUnavailable(msg) from None
        for sensor in sensors:
            self._check_temperature(sensor)

    def _check_temperature(self, sensor, device=None):
        """Warn about a sensor exceeding its threshold or a misconfigured maximum."""
        if self.max_temperature_celsius is not None:
            location = f"{device} " if device else ""
            if self.max_temperature_celsius is SENSOR_HIGH_THRESHOLD:
                threshold = sensor.high
            else:
                threshold = self.max_temperature_celsius
                if sensor.critical is not None and threshold > sensor.critical:
                    msg = (
                        f"{location}Configured maximum {threshold:.1f}\u202f°C exceeds "
                        f"critical threshold {sensor.critical:.1f}\u202f°C of sensor {sensor.label!r}"
                    )
                    raise ServiceWarning(msg)
            if threshold is not None and sensor.current >= threshold:
                msg = f"{location}{sensor.label} {sensor.current:.1f}\u202f°C"
                raise ServiceWarning(msg)

3rd Party Services

To use these checks, install the dependencies and configure them:

pip install django-health-check[redis,rabbitmq,celery,kafka,crontask,threadmill]

health_check.contrib.celery.Ping dataclass

Bases: HealthCheck

Check Celery worker availability using the ping control command.

Parameters:

Name	Type	Description	Default
app	Celery	Celery application instance to use for the health check, defaults to the default Celery app.	app_or_default()
timeout	timedelta	Timeout duration for the ping command.	timedelta(seconds=1)
limit	int | None	Maximum number of workers to wait for before returning. If not specified, waits for the full timeout duration. When set, returns immediately after receiving responses from this many workers.	None

Source code in health_check/contrib/celery.py

@dataclasses.dataclass
class Ping(HealthCheck):
    """
    Check Celery worker availability using the ping control command.

    Args:
        app: Celery application instance to use for the health check, defaults to the [default Celery app][celery.app.default_app].
        timeout: Timeout duration for the ping command.
        limit: Maximum number of workers to wait for before returning. If not
            specified, waits for the full timeout duration. When set, returns
            immediately after receiving responses from this many workers.

    """

    CORRECT_PING_RESPONSE: typing.ClassVar[dict[str, str]] = {"ok": "pong"}
    app: celery.Celery = dataclasses.field(default_factory=app_or_default)
    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=1), repr=False
    )
    limit: int | None = dataclasses.field(default=None, repr=False)

    def run(self):
        try:
            ping_result = self.app.control.ping(
                timeout=self.timeout.total_seconds(), limit=self.limit
            )
        except OSError as e:
            raise ServiceUnavailable("IOError") from e
        except NotImplementedError as e:
            msg = "NotImplementedError: Make sure CELERY_RESULT_BACKEND is set"
            raise ServiceUnavailable(msg) from e
        else:
            if not ping_result:
                msg = "Celery workers unavailable"
                raise ServiceUnavailable(msg)
            self.check_active_queues(*self.active_workers(ping_result))

    def active_workers(self, ping_result):
        for result in ping_result:
            worker, response = list(result.items())[0]
            if response != self.CORRECT_PING_RESPONSE:
                msg = f"Celery worker {worker} response was incorrect"
                raise ServiceUnavailable(msg)
            yield worker

    def check_active_queues(self, *active_workers):
        try:
            defined_queues = {queue.name for queue in self.app.conf.task_queues}
        except TypeError:
            # conf.task_queues may be None
            defined_queues = {self.app.conf.task_default_queue}
        active_queues = {
            queue.get("name")
            for queues in (
                self.app.control.inspect(active_workers).active_queues() or {}
            ).values()
            for queue in queues
        }

        for queue in defined_queues - active_queues:
            msg = f"No worker for Celery task queue {queue}"
            raise ServiceUnavailable(msg)

health_check.contrib.crontask.Scheduler dataclass

Bases: HealthCheck

Presence of the django-crontask scheduler operating.

Crontask renews its lock as a dead man's switch while it runs, so a missing key means it is dead. Requires CRONTASK["REDIS_URL"], since crontask uses a no-op lock without it.

Parameters:

Name	Type	Description	Default
client_factory	Callable[[], Redis]	A callable returning a Redis client, defaults to create_client_from_settings.	create_client_from_settings

Source code in health_check/contrib/crontask.py

@dataclasses.dataclass
class Scheduler(HealthCheck):
    """
    Presence of the [django-crontask] scheduler operating.

    Crontask renews its lock as a dead man's switch while it runs, so a
    missing key means it is dead. Requires `CRONTASK["REDIS_URL"]`, since
    crontask uses a no-op lock without it.

    [django-crontask]: https://github.com/codingjoe/django-crontask

    Args:
        client_factory: A callable returning a Redis client, defaults to
            `create_client_from_settings`.

    """

    client_factory: typing.Callable[[], RedisClient] = dataclasses.field(
        default=create_client_from_settings, repr=False
    )

    async def run(self):
        client = self.client_factory()
        try:
            # django-crontask hardcodes this key with no setting; update if renamed
            if not await client.exists("crontask-lock"):
                msg = "No django-crontask scheduler is running."
                raise ServiceUnavailable(msg)
        except exceptions.RedisError as e:
            msg = "Unable to query the crontask lock."
            raise ServiceUnavailable(msg) from e
        finally:
            await client.aclose()

health_check.contrib.kafka.Kafka dataclass

Bases: HealthCheck

Check Kafka service by connecting to a Kafka broker and listing topics.

Parameters:

Name	Type	Description	Default
bootstrap_servers	list[str]	List of Kafka bootstrap servers, e.g., ['localhost:9092'].	required
timeout	timedelta	Timeout duration for the connection check as a datetime.timedelta.	timedelta(seconds=10)

Source code in health_check/contrib/kafka.py

@dataclasses.dataclass
class Kafka(HealthCheck):
    """
    Check Kafka service by connecting to a Kafka broker and listing topics.

    Args:
        bootstrap_servers: List of Kafka bootstrap servers, e.g., ['localhost:9092'].
        timeout: Timeout duration for the connection check as a datetime.timedelta.

    """

    bootstrap_servers: list[str]
    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )

    async def run(self):
        logger.debug(
            "Connecting to Kafka bootstrap servers %r ...",
            self.bootstrap_servers,
        )

        # Create a consumer with minimal configuration for health check
        timeout_ms = int(self.timeout.total_seconds() * 1000)
        consumer = AIOConsumer(
            {
                "bootstrap.servers": ",".join(self.bootstrap_servers),
                "client.id": "health-check",
                "group.id": "health-check",
                "session.timeout.ms": timeout_ms,
                "socket.timeout.ms": timeout_ms,
            }
        )

        try:
            if not (
                (
                    cluster_metadata := await consumer.list_topics(
                        timeout=self.timeout.total_seconds()
                    )
                )
                and cluster_metadata.topics
            ):
                msg = "Failed to retrieve Kafka topics."
                raise ServiceUnavailable(msg)

        except KafkaException as e:
            msg = "Unable to connect"
            raise ServiceUnavailable(msg) from e
        else:
            logger.debug(
                "Connection established. Kafka is healthy. Found %d topics.",
                len(cluster_metadata.topics),
            )
        finally:
            await consumer.close()

health_check.contrib.rabbitmq.RabbitMQ dataclass

Bases: HealthCheck

Check RabbitMQ service by opening and closing a broker channel.

Parameters:

Name	Type	Description	Default
amqp_url	str	The URL of the RabbitMQ broker to connect to, e.g., 'amqp://guest:guest@localhost:5672//'.	required
timeout	timedelta	Timeout for the connection attempt.	timedelta(seconds=5)

Source code in health_check/contrib/rabbitmq.py

@dataclasses.dataclass
class RabbitMQ(HealthCheck):
    """
    Check RabbitMQ service by opening and closing a broker channel.

    Args:
        amqp_url (str): The URL of the RabbitMQ broker to connect to, e.g., 'amqp://guest:guest@localhost:5672//'.
        timeout: Timeout for the connection attempt.

    """

    amqp_url: str = dataclasses.field(repr=False)
    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=5), repr=False
    )

    def __repr__(self) -> str:
        arguments = ", ".join(
            f"{key}={value!r}" for key, value in self._connection_details().items()
        )
        return f"{self.__class__.__name__}({arguments})"

    @property
    def labels(self) -> dict[str, str]:
        return super().labels | {
            key: str(value) for key, value in self._connection_details().items()
        }

    def _connection_details(self) -> dict[str, str | int]:
        try:
            url = urllib.parse.urlsplit(self.amqp_url)
            details = {"scheme": url.scheme, "host": url.hostname, "port": url.port}
        except ValueError:
            return {}
        if not details["host"]:
            return {}
        return {key: value for key, value in details.items() if value}

    async def run(self):
        logger.debug("Attempting to connect to %r...", self)
        try:
            # conn is used as a context to release opened resources later
            connection = await aio_pika.connect_robust(
                self.amqp_url, timeout=self.timeout.total_seconds()
            )
            await connection.close()
        except ConnectionRefusedError as e:
            msg = "Unable to connect to RabbitMQ: Connection was refused."
            raise ServiceUnavailable(msg) from e
        except aio_pika.exceptions.ProbableAuthenticationError as e:
            msg = "Unable to connect to RabbitMQ: Authentication error."
            raise ServiceUnavailable(msg) from e
        except OSError as e:
            raise ServiceUnavailable("IOError") from e
        else:
            logger.debug("Connection established. RabbitMQ is healthy.")

health_check.contrib.redis.Redis dataclass

Bases: HealthCheck

Check Redis service by pinging a Redis client.

This check works with any Redis client that implements the ping() method, including standard Redis, Sentinel, and Cluster clients.

Parameters:

Name	Type	Description	Default
client_factory	Callable[[], Redis | RedisCluster] | None	A callable that returns an instance of a Redis client.	None
client	Redis | RedisCluster | None	Deprecated, use client_factory instead.	None

Examples:

Using a standard Redis client:

>>> from redis.asyncio import Redis as RedisClient
>>> Redis(client_factory=lambda: RedisClient(host='localhost', port=6379))

Using from_url to create a client:

>>> from redis.asyncio import Redis as RedisClient
>>> Redis(client_factory=lambda: RedisClient.from_url('redis://localhost:6379'))

Using a Cluster client:

>>> from redis.asyncio import RedisCluster
>>> Redis(client_factory=lambda: RedisCluster(host='localhost', port=7000))

Using a Sentinel client:

>>> from redis.asyncio import Sentinel
>>> Redis(client_factory=lambda: Sentinel([('localhost', 26379)]).master_for('mymaster'))

Source code in health_check/contrib/redis.py

@dataclasses.dataclass
class Redis(HealthCheck):
    """
    Check Redis service by pinging a Redis client.

    This check works with any Redis client that implements the ping() method,
    including standard Redis, Sentinel, and Cluster clients.

    Args:
        client_factory: A callable that returns an instance of a Redis client.
        client: Deprecated, use `client_factory` instead.

    Examples:
        Using a standard Redis client:
        >>> from redis.asyncio import Redis as RedisClient
        >>> Redis(client_factory=lambda: RedisClient(host='localhost', port=6379))

        Using from_url to create a client:
        >>> from redis.asyncio import Redis as RedisClient
        >>> Redis(client_factory=lambda: RedisClient.from_url('redis://localhost:6379'))

        Using a Cluster client:
        >>> from redis.asyncio import RedisCluster
        >>> Redis(client_factory=lambda: RedisCluster(host='localhost', port=7000))

        Using a Sentinel client:
        >>> from redis.asyncio import Sentinel
        >>> Redis(client_factory=lambda: Sentinel([('localhost', 26379)]).master_for('mymaster'))

    """

    client: RedisClient | RedisCluster | None = dataclasses.field(
        repr=False, default=None
    )
    client_factory: typing.Callable[[], RedisClient | RedisCluster] | None = (
        dataclasses.field(repr=False, default=None)
    )

    def _connection_kwargs(self) -> dict[str, object]:
        """Return identifying Redis connection parameters for repr and labels."""
        client = (
            self.client_factory() if self.client_factory is not None else self.client
        )
        try:
            return {
                key: value
                for key, value in sorted(
                    client.connection_pool.connection_kwargs.items()
                )
                if key in {"host", "port", "db"}
            }
        except AttributeError:
            pass
        try:
            return {"hosts": [node.name for node in client.startup_nodes]}
        except AttributeError:
            return {}

    def __repr__(self):
        match self._connection_kwargs():
            case {"hosts": hosts}:
                return f"Redis(client=RedisCluster(hosts={hosts!r}))"
            case kwargs if kwargs:
                return f"Redis({', '.join(f'{key}={value!r}' for key, value in kwargs.items())})"
            case _:
                return super().__repr__()

    @property
    def labels(self) -> dict[str, str]:
        return super().labels | {
            key: str(value) for key, value in self._connection_kwargs().items()
        }

    def __post_init__(self):
        # Validate that exactly one of client or client_factory is provided
        if self.client is not None and self.client_factory is not None:
            msg = "Provide exactly one of `client` or `client_factory`, not both."
            raise ValueError(msg)
        if self.client is None and self.client_factory is None:
            msg = (
                "You must provide either `client` (deprecated) or `client_factory` "
                "when instantiating `Redis`."
            )
            raise ValueError(msg)

        # Emit deprecation warning if using the old client parameter
        if self.client is not None:
            warnings.warn(
                "The `client` argument is deprecated and will be removed in a future version. "
                "Please use `client_factory` instead.",
                DeprecationWarning,
                stacklevel=2,
            )

    async def run(self):
        # Create a new client for this health check request
        if self.client_factory is not None:
            client = self.client_factory()
            should_close = True
        else:
            # Use the deprecated client parameter (user manages lifecycle)
            client = self.client
            should_close = False

        logger.debug("Pinging Redis client...")
        try:
            await client.ping()
        except ConnectionRefusedError as e:
            msg = "Unable to connect to Redis: Connection was refused."
            raise ServiceUnavailable(msg) from e
        except exceptions.TimeoutError as e:
            msg = "Unable to connect to Redis: Timeout."
            raise ServiceUnavailable(msg) from e
        except exceptions.ConnectionError as e:
            msg = "Unable to connect to Redis: Connection Error"
            raise ServiceUnavailable(msg) from e
        else:
            logger.debug("Connection established. Redis is healthy.")
        finally:
            # Only close clients created by client_factory
            if should_close:
                await client.aclose()

health_check.contrib.threadmill.Threadmill dataclass

Bases: HealthCheck

Monitor Threadmill, the task backend for Django's task framework.

The check reports the number of ready and deferred tasks and the number of failed tasks of a single queue. If the task broker does not answer, the check fails. It only reads the telemetry, therefore it cannot tell if a worker is alive. If workers stop, the number of ready tasks increases.

Parameters:

Name	Type	Description	Default
alias	str	Alias of the TASKS backend.	DEFAULT_TASK_BACKEND_ALIAS
queue_name	str	Queue for the telemetry.	DEFAULT_TASK_QUEUE_NAME
max_pending_tasks	int | None	Maximum number of ready and deferred tasks before the check fails, or None to disable the limit.	1000
max_failed_tasks	int | None	Maximum number of failed tasks before the check warns, or None to disable the limit.	None
timeout	timedelta	Timeout for the telemetry request.	timedelta(seconds=5)

Source code in health_check/contrib/threadmill.py

@dataclasses.dataclass
class Threadmill(HealthCheck):
    """
    Monitor [Threadmill], the task backend for [Django's task framework].

    The check reports the number of ready and deferred tasks and the number of
    failed tasks of a single queue. If the task broker does not answer, the check
    fails. It only reads the telemetry, therefore it cannot tell if a worker is
    alive. If workers stop, the number of ready tasks increases.

    [Threadmill]: https://github.com/codingjoe/threadmill
    [Django's task framework]: https://docs.djangoproject.com/en/stable/topics/tasks/

    Args:
        alias: Alias of the `TASKS` backend.
        queue_name: Queue for the telemetry.
        max_pending_tasks: Maximum number of ready and deferred tasks before the check fails, or `None` to disable the limit.
        max_failed_tasks: Maximum number of failed tasks before the check warns, or `None` to disable the limit.
        timeout: Timeout for the telemetry request.

    """

    alias: str = DEFAULT_TASK_BACKEND_ALIAS
    queue_name: str = DEFAULT_TASK_QUEUE_NAME
    max_pending_tasks: int | None = dataclasses.field(default=1000, repr=False)
    max_failed_tasks: int | None = dataclasses.field(default=None, repr=False)
    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=5), repr=False
    )

    async def run(self) -> None:
        try:
            backend = task_backends[self.alias]
        except InvalidTaskBackend as e:
            msg = "Task backend alias does not exist"
            raise ServiceUnavailable(msg) from e
        if not isinstance(backend, ThreadmillTaskBackend):
            msg = "Task backend does not support queue telemetry"
            raise ServiceUnavailable(msg)
        if self.queue_name not in backend.queues:
            msg = "Task queue does not exist"
            raise ServiceUnavailable(msg)
        # The handler caches a backend bound to the event loop that created it.
        # Build a new backend for each probe; threadmill manages its client.
        probe = task_backends.create_connection(self.alias)
        try:
            telemetry = await asyncio.wait_for(
                probe.queue_stats(), timeout=self.timeout.total_seconds()
            )
        except TimeoutError as e:
            msg = "Threadmill queue telemetry timed out"
            raise ServiceUnavailable(msg) from e
        except (RedisError, OSError) as e:
            msg = "Unable to connect to threadmill task broker"
            raise ServiceUnavailable(msg) from e
        counts = telemetry.queues[self.queue_name].counts
        pending = counts.ready + counts.deferred
        failed = counts.failed
        if self.max_pending_tasks is not None and pending > self.max_pending_tasks:
            msg = f"{pending} pending tasks exceed max_pending_tasks of {self.max_pending_tasks}"
            raise ServiceUnavailable(msg)
        if self.max_failed_tasks is not None and failed > self.max_failed_tasks:
            msg = f"{failed} failed tasks exceed max_failed_tasks of {self.max_failed_tasks}"
            raise ServiceWarning(msg)

Cloud Provider Status

Watch cloud provider health through their public RSS/Atom status feeds or APIs.

Each cloud provider check requires an extra. The extra depends on the provider:

pip install django-health-check[rss,atlassian]

health_check.contrib.rss.AWS dataclass

Bases: Feed

Check AWS service status via their public RSS status feeds.

Parameters:

Name	Type	Description	Default
region	str	AWS region code (e.g., 'us-east-1', 'eu-west-1').	required
service	str	AWS service name (e.g., 'ec2', 's3', 'rds').	required
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
max_age	timedelta	Maximum age for an incident to be considered active.	timedelta(hours=8)

Source code in health_check/contrib/rss.py

@dataclasses.dataclass
class AWS(Feed):
    """
    Check AWS service status via their public RSS status feeds.

    Args:
        region: AWS region code (e.g., 'us-east-1', 'eu-west-1').
        service: AWS service name (e.g., 'ec2', 's3', 'rds').
        timeout: Request timeout duration.
        max_age: Maximum age for an incident to be considered active.

    """

    region: str
    service: str
    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    max_age: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(hours=8), repr=False
    )

    def __post_init__(self):
        self.feed_url: str = (
            f"https://status.aws.amazon.com/rss/{self.service}-{self.region}.rss"
        )

health_check.contrib.rss.Azure dataclass

Bases: Feed

Check Azure platform status via their public RSS status feed.

Parameters:

Name	Type	Description	Default
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
max_age	timedelta	Maximum age for an incident to be considered active.	timedelta(hours=8)

Source code in health_check/contrib/rss.py

@dataclasses.dataclass
class Azure(Feed):
    """
    Check Azure platform status via their public RSS status feed.

    Args:
        timeout: Request timeout duration.
        max_age: Maximum age for an incident to be considered active.

    """

    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    max_age: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(hours=8), repr=False
    )
    feed_url: str = dataclasses.field(
        default="https://rssfeed.azure.status.microsoft/en-us/status/feed/",
        init=False,
        repr=False,
    )

health_check.contrib.atlassian.Cloudflare dataclass

Bases: AtlassianStatusPage

Check Cloudflare platform status via Atlassian Status Page API v2.

Parameters:

Name	Type	Description	Default
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
component	str	Name of a specific component to monitor. Monitors all components when empty.	''

Source code in health_check/contrib/atlassian.py

@dataclasses.dataclass
class Cloudflare(AtlassianStatusPage):
    """
    Check Cloudflare platform status via Atlassian Status Page API v2.

    Args:
        timeout: Request timeout duration.
        component: Name of a specific component to monitor. Monitors all
            components when empty.

    """

    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    base_url: str = dataclasses.field(
        default="https://www.cloudflarestatus.com", init=False, repr=False
    )
    component: str = ""

health_check.contrib.atlassian.DigitalOcean dataclass

Bases: AtlassianStatusPage

Check DigitalOcean platform status via Atlassian Status Page API v2.

Parameters:

Name	Type	Description	Default
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
component	str	Name of a specific component to monitor. Monitors all components when empty.	''

Source code in health_check/contrib/atlassian.py

@dataclasses.dataclass
class DigitalOcean(AtlassianStatusPage):
    """
    Check DigitalOcean platform status via Atlassian Status Page API v2.

    Args:
        timeout: Request timeout duration.
        component: Name of a specific component to monitor. Monitors all
            components when empty.

    """

    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    base_url: str = dataclasses.field(
        default="https://status.digitalocean.com", init=False, repr=False
    )
    component: str = ""

health_check.contrib.atlassian.FlyIo dataclass

Bases: AtlassianStatusPage

Check Fly.io platform status via Atlassian Status Page API v2.

Parameters:

Name	Type	Description	Default
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
component	str	Name of a specific component to monitor. Monitors all components when empty.	''

Source code in health_check/contrib/atlassian.py

@dataclasses.dataclass
class FlyIo(AtlassianStatusPage):
    """
    Check Fly.io platform status via Atlassian Status Page API v2.

    Args:
        timeout: Request timeout duration.
        component: Name of a specific component to monitor. Monitors all
            components when empty.

    """

    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    base_url: str = dataclasses.field(
        default="https://status.flyio.net", init=False, repr=False
    )
    component: str = ""

health_check.contrib.atlassian.GitHub dataclass

Bases: AtlassianStatusPage

Check GitHub platform status via Atlassian Status Page API v2.

Parameters:

Name	Type	Description	Default
enterprise_region	EnterpriseRegion | None	GitHub Enterprise status page region (if applicable).	None
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
component	str	Name of a specific component to monitor. Monitors all components when empty.	''

Source code in health_check/contrib/atlassian.py

@dataclasses.dataclass
class GitHub(AtlassianStatusPage):
    """
    Check GitHub platform status via Atlassian Status Page API v2.

    Args:
        enterprise_region: GitHub Enterprise status page region (if applicable).
        timeout: Request timeout duration.
        component: Name of a specific component to monitor. Monitors all
            components when empty.

    """

    class EnterpriseRegion(enum.StrEnum):
        """GitHub Enterprise status page regions."""

        australia = "au"
        """Australia."""
        eu = "eu"
        """Europe."""
        japan = "jp"
        """Japan."""
        us = "us"
        """United States."""

    enterprise_region: EnterpriseRegion | None = None
    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    component: str = ""

    def __post_init__(self):
        self.base_url = f"https://{self.enterprise_region if self.enterprise_region else 'www'}.githubstatus.com"

EnterpriseRegion

Bases: StrEnum

GitHub Enterprise status page regions.

Source code in health_check/contrib/atlassian.py

class EnterpriseRegion(enum.StrEnum):
    """GitHub Enterprise status page regions."""

    australia = "au"
    """Australia."""
    eu = "eu"
    """Europe."""
    japan = "jp"
    """Japan."""
    us = "us"
    """United States."""

australia = 'au' class-attribute instance-attribute

Australia.

eu = 'eu' class-attribute instance-attribute

Europe.

japan = 'jp' class-attribute instance-attribute

Japan.

us = 'us' class-attribute instance-attribute

United States.

health_check.contrib.rss.GoogleCloud dataclass

Bases: Feed

Check Google Cloud platform status via their public Atom status feed.

Parameters:

Name	Type	Description	Default
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
max_age	timedelta	Maximum age for an incident to be considered active.	timedelta(hours=8)

Source code in health_check/contrib/rss.py

@dataclasses.dataclass
class GoogleCloud(Feed):
    """
    Check Google Cloud platform status via their public Atom status feed.

    Args:
        timeout: Request timeout duration.
        max_age: Maximum age for an incident to be considered active.

    """

    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    max_age: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(hours=8), repr=False
    )

    feed_url: str = dataclasses.field(
        default="https://status.cloud.google.com/en/feed.atom", init=False, repr=False
    )

health_check.contrib.rss.Heroku dataclass

Bases: Feed

Check Heroku platform status via their public RSS status feed.

Parameters:

Name	Type	Description	Default
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
max_age	timedelta	Maximum age for an incident to be considered active.	timedelta(hours=8)

Source code in health_check/contrib/rss.py

@dataclasses.dataclass
class Heroku(Feed):
    """
    Check Heroku platform status via their public RSS status feed.

    Args:
        timeout: Request timeout duration.
        max_age: Maximum age for an incident to be considered active.

    """

    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    max_age: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(hours=8), repr=False
    )
    feed_url: str = dataclasses.field(
        default="https://status.heroku.com/feed", init=False, repr=False
    )

health_check.contrib.rss.Hetzner dataclass

Bases: Feed

Check Hetzner platform status via their public ATOM status feed.

Parameters:

Name	Type	Description	Default
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
max_age	timedelta	Maximum age for an incident to be considered active.	timedelta(hours=8)

Source code in health_check/contrib/rss.py

@dataclasses.dataclass
class Hetzner(Feed):
    """
    Check Hetzner platform status via their public ATOM status feed.

    Args:
        timeout: Request timeout duration.
        max_age: Maximum age for an incident to be considered active.

    """

    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    max_age: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(hours=8), repr=False
    )
    feed_url: str = dataclasses.field(
        default="https://status.hetzner.com/en.atom", init=False, repr=False
    )

health_check.contrib.atlassian.Npm dataclass

Bases: AtlassianStatusPage

Health check for npm registry status via Atlassian Status Page API v2.

Parameters:

Name	Type	Description	Default
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
component	str	Name of a specific component to monitor. Monitors all components when empty.	'Package installation'

Source code in health_check/contrib/atlassian.py

@dataclasses.dataclass
class Npm(AtlassianStatusPage):
    """
    Health check for npm registry status via Atlassian Status Page API v2.

    Args:
        timeout: Request timeout duration.
        component: Name of a specific component to monitor. Monitors all
            components when empty.

    """

    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    base_url: str = dataclasses.field(
        default="https://status.npmjs.org", init=False, repr=False
    )
    component: str = "Package installation"

health_check.contrib.atlassian.PlatformSh dataclass

Bases: AtlassianStatusPage

Check Platform.sh platform status via Atlassian Status Page API v2.

Parameters:

Name	Type	Description	Default
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
component	str	Name of a specific component to monitor. Monitors all components when empty.	''

Source code in health_check/contrib/atlassian.py

@dataclasses.dataclass
class PlatformSh(AtlassianStatusPage):
    """
    Check Platform.sh platform status via Atlassian Status Page API v2.

    Args:
        timeout: Request timeout duration.
        component: Name of a specific component to monitor. Monitors all
            components when empty.

    """

    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    base_url: str = dataclasses.field(
        default="https://status.platform.sh", init=False, repr=False
    )
    component: str = ""

health_check.contrib.atlassian.PyPI dataclass

Bases: AtlassianStatusPage

Check Python Package Index (PyPI) platform status via Atlassian Status Page API v2.

Parameters:

Name	Type	Description	Default
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
component	str	Name of a component or component group to monitor. Monitors all components when empty.	'PyPI'

Source code in health_check/contrib/atlassian.py

@dataclasses.dataclass
class PyPI(AtlassianStatusPage):
    """
    Check Python Package Index (PyPI) platform status via Atlassian Status Page API v2.

    Args:
        timeout: Request timeout duration.
        component: Name of a component or component group to monitor. Monitors
            all components when empty.

    """

    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    base_url: str = dataclasses.field(
        default="https://status.python.org", init=False, repr=False
    )
    component: str = "PyPI"

health_check.contrib.atlassian.Render dataclass

Bases: AtlassianStatusPage

Check Render platform status via Atlassian Status Page API v2.

Parameters:

Name	Type	Description	Default
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
component	str	Name of a specific component to monitor. Monitors all components when empty.	''

Source code in health_check/contrib/atlassian.py

@dataclasses.dataclass
class Render(AtlassianStatusPage):
    """
    Check Render platform status via Atlassian Status Page API v2.

    Args:
        timeout: Request timeout duration.
        component: Name of a specific component to monitor. Monitors all
            components when empty.

    """

    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    base_url: str = dataclasses.field(
        default="https://status.render.com", init=False, repr=False
    )
    component: str = ""

health_check.contrib.atlassian.Sentry dataclass

Bases: AtlassianStatusPage

Check Sentry platform status via Atlassian Status Page API v2.

Parameters:

Name	Type	Description	Default
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
component	str	Name of a specific component to monitor. Monitors all components when empty.	''

Source code in health_check/contrib/atlassian.py

@dataclasses.dataclass
class Sentry(AtlassianStatusPage):
    """
    Check Sentry platform status via Atlassian Status Page API v2.

    Args:
        timeout: Request timeout duration.
        component: Name of a specific component to monitor. Monitors all
            components when empty.

    """

    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    base_url: str = dataclasses.field(
        default="https://status.sentry.io", init=False, repr=False
    )
    component: str = ""

health_check.contrib.atlassian.Vercel dataclass

Bases: AtlassianStatusPage

Check Vercel platform status via Atlassian Status Page API v2.

Parameters:

Name	Type	Description	Default
timeout	timedelta	Request timeout duration.	timedelta(seconds=10)
component	str	Name of a specific component to monitor. Monitors all components when empty.	''

Source code in health_check/contrib/atlassian.py

@dataclasses.dataclass
class Vercel(AtlassianStatusPage):
    """
    Check Vercel platform status via Atlassian Status Page API v2.

    Args:
        timeout: Request timeout duration.
        component: Name of a specific component to monitor. Monitors all
            components when empty.

    """

    timeout: datetime.timedelta = dataclasses.field(
        default=datetime.timedelta(seconds=10), repr=False
    )
    base_url: str = dataclasses.field(
        default="https://www.vercel-status.com", init=False, repr=False
    )
    component: str = ""

Custom Status Page Feeds

Write your own status page proxy checks with these classes. Subclasses must implement the documented attributes.

health_check.contrib.rss.Feed dataclass

Bases: HealthCheck

Base class for cloud provider status feed health checks.

Monitor cloud provider service health via their public RSS or Atom status feeds.

Subclasses must provide:

Attributes:

Name	Type	Description
feed_url	str	The full URL of the RSS or Atom feed to monitor.
timeout	timedelta	Maximum duration to wait for the HTTP request before failing.
max_age	timedelta	Maximum age for an incident entry to be considered active.

The timeout and max_age values are used to control how long the health check waits for the feed and how far back in time incidents are considered relevant. The feed_url is used to fetch the status feed.

Examples:

>>> import dataclasses
>>> import datetime
>>> import typing
>>> from health_check.contrib.rss import Feed
>>> @dataclasses.dataclass
... class MyProviderStatus(Feed):
...     """Check MyProvider status via its public RSS feed."""
...     timeout: datetime.timedelta = dataclasses.field(
...         default=datetime.timedelta(seconds=10),
...         repr=False,
...     )
...     max_age: datetime.timedelta = dataclasses.field(
...         default=datetime.timedelta(hours=4),
...         repr=False,
...     )
...     feed_url: typing.ClassVar[str] = "https://status.myprovider.com/feed"

Source code in health_check/contrib/rss.py

class Feed(HealthCheck):
    """
    Base class for cloud provider status feed health checks.

    Monitor cloud provider service health via their public RSS or Atom status feeds.

    Subclasses must provide:

    Attributes:
        feed_url: The full URL of the RSS or Atom feed to monitor.
        timeout: Maximum duration to wait for the HTTP request before failing.
        max_age: Maximum age for an incident entry to be considered active.

    The `timeout` and `max_age` values are used to control how long the
    health check waits for the feed and how far back in time incidents are
    considered relevant. The `feed_url` is used to fetch the status feed.

    Examples:
        >>> import dataclasses
        >>> import datetime
        >>> import typing
        >>> from health_check.contrib.rss import Feed
        >>> @dataclasses.dataclass
        ... class MyProviderStatus(Feed):
        ...     \"\"\"Check MyProvider status via its public RSS feed.\"\"\"
        ...     timeout: datetime.timedelta = dataclasses.field(
        ...         default=datetime.timedelta(seconds=10),
        ...         repr=False,
        ...     )
        ...     max_age: datetime.timedelta = dataclasses.field(
        ...         default=datetime.timedelta(hours=4),
        ...         repr=False,
        ...     )
        ...     feed_url: typing.ClassVar[str] = "https://status.myprovider.com/feed"

    """

    feed_url: str = NotImplemented
    timeout: datetime.timedelta = NotImplemented
    max_age: datetime.timedelta = NotImplemented

    async def run(self):
        logger.debug("Fetching feed from %s", self.feed_url)

        async with httpx.AsyncClient() as client:
            try:
                response = await client.get(
                    self.feed_url,
                    headers={"User-Agent": f"django-health-check@{__version__}"},
                    timeout=self.timeout.total_seconds(),
                    follow_redirects=True,
                )
            except httpx.TimeoutException as e:
                msg = "Feed request timed out"
                raise ServiceUnavailable(msg) from e
            except httpx.RequestError as e:
                msg = f"Failed to fetch feed: {e}"
                raise ServiceUnavailable(msg) from e

            try:
                response.raise_for_status()
            except httpx.HTTPStatusError as e:
                msg = f"HTTP error {e.response.status_code} fetching feed from {self.feed_url!r}"
                raise ServiceUnavailable(msg) from e

            content = response.text

        feed = feedparser.parse(content)

        if feed.bozo:
            # feedparser sets bozo=1 for malformed feeds
            logger.warning("Feed parsing encountered errors: %s", feed.bozo_exception)

        if not feed.entries:
            logger.debug("No entries found in feed")
            return

        if incidents := list(self._recent_incidents(feed.entries)):
            raise StatusPageWarning(
                "\n".join(
                    f"{getattr(entry, 'title', 'Unknown Incident') or 'Unknown Incident'}:"
                    f" {getattr(entry, 'link', self.feed_url) or self.feed_url}"
                    for entry, _ in incidents
                ),
                timestamp=max(
                    filter(None, (date for _, date in incidents)), default=None
                ),
            )

        logger.debug("No recent incidents found in feed")

    def _recent_incidents(self, entries):
        """Yield recent (entry, timestamp) pairs from feed entries."""
        for entry in entries:
            date = self._extract_date(entry)
            if date is None or self._is_date_recent(date):
                yield entry, date

    def _is_date_recent(self, date):
        """Check if a timestamp falls within the configured max_age window."""
        now = datetime.datetime.now(tz=datetime.UTC)
        return now >= date > now - self.max_age

    def _extract_date(self, entry):
        # feedparser normalizes both RSS and Atom dates to struct_time
        # Try published first, then updated
        for date_field in ["published_parsed", "updated_parsed"]:
            if date_tuple := getattr(entry, date_field, None):
                try:
                    # Convert struct_time to datetime
                    return datetime.datetime(*date_tuple[:6], tzinfo=datetime.UTC)
                except ValueError, TypeError:
                    logger.warning(
                        "Failed to parse date from entry %r for %r",
                        date_tuple,
                        self.feed_url,
                        exc_info=True,
                    )
        return None

health_check.contrib.atlassian.AtlassianStatusPage dataclass

Bases: HealthCheck

Base class for Atlassian status page health checks.

Monitor cloud provider service health via Atlassian Status Page API v2.

Each subclass should define the base_url for the specific status page and appropriate timeout value.

When component is non-empty, only incidents affecting that named component are reported. Naming a component group also reports incidents of its child components. Use separate check instances to monitor multiple components independently.

Examples:

>>> import dataclasses
>>> import datetime
>>> from health_check.contrib.atlassian import AtlassianStatusPage
>>> @dataclasses.dataclass
... class FlyIo(AtlassianStatusPage):
...     timeout: datetime.timedelta = datetime.timedelta(seconds=10)
...     base_url: str = dataclasses.field(default="https://status.flyio.net", init=False, repr=False)

Source code in health_check/contrib/atlassian.py

class AtlassianStatusPage(HealthCheck):
    """
    Base class for Atlassian status page health checks.

    Monitor cloud provider service health via Atlassian Status Page API v2.

    Each subclass should define the `base_url` for the specific status page
    and appropriate `timeout` value.

    When `component` is non-empty, only incidents affecting that named component are
    reported. Naming a component group also reports incidents of its child components.
    Use separate check instances to monitor multiple components independently.

    Examples:
        >>> import dataclasses
        >>> import datetime
        >>> from health_check.contrib.atlassian import AtlassianStatusPage
        >>> @dataclasses.dataclass
        ... class FlyIo(AtlassianStatusPage):
        ...     timeout: datetime.timedelta = datetime.timedelta(seconds=10)
        ...     base_url: str = dataclasses.field(default="https://status.flyio.net", init=False, repr=False)

    """

    base_url: str = NotImplemented
    timeout: datetime.timedelta = NotImplemented
    component: str = ""

    def _watched_component_names(self, components):
        """Yield the configured component name and the names of its child components."""
        if self.component:
            try:
                watched = next(c for c in components if c["name"] == self.component)
            except StopIteration as e:
                msg = f"Component {self.component!r} not found"
                raise ServiceReturnedUnexpectedResult(msg) from e
            yield self.component
            yield from (
                c["name"]
                for c in components
                if (group_id := watched.get("id")) is not None
                and c.get("group_id") == group_id
            )

    async def run(self):
        if incidents := [i async for i in self._fetch_incidents()]:
            raise StatusPageWarning(
                "\n".join(msg for msg, _ in incidents),
                timestamp=max(ts for _, ts in incidents),
            )
        logger.debug("No incidents found")

    async def _fetch_incidents(self):
        api_url = f"{self.base_url}/api/v2/summary.json"
        logger.debug("Fetching incidents from %r", api_url)

        async with httpx.AsyncClient() as client:
            try:
                response = await client.get(
                    api_url,
                    headers={"User-Agent": f"django-health-check@{__version__}"},
                    timeout=self.timeout.total_seconds(),
                    follow_redirects=True,
                )
            except httpx.TimeoutException as e:
                msg = "API request timed out"
                raise ServiceUnavailable(msg) from e
            except httpx.RequestError as e:
                msg = f"Failed to fetch API: {e}"
                raise ServiceUnavailable(msg) from e

            try:
                response.raise_for_status()
            except httpx.HTTPStatusError as e:
                msg = (
                    f"HTTP error {e.response.status_code} fetching API from {api_url!r}"
                )
                raise ServiceUnavailable(msg) from e

            try:
                data = response.json()
            except ValueError as e:
                msg = "Failed to parse JSON response"
                raise ServiceUnavailable(msg) from e

        watched_names = set(self._watched_component_names(data.get("components", ())))
        try:
            for incident in data["incidents"]:
                if (incident.get("status") not in ("resolved", "postmortem")) and (
                    not watched_names
                    or any(c["name"] in watched_names for c in incident["components"])
                ):
                    yield (
                        f"{incident['name']}: {incident['shortlink']}",
                        datetime.datetime.fromisoformat(incident["updated_at"]),
                    )
        except KeyError as e:
            msg = "Unexpected API response structure"
            raise ServiceReturnedUnexpectedResult(msg) from e