Skip to content

FastHealthcheck

Framework agnostic health checks with integrations for most popular ASGI frameworks: FastAPI / Faststream / Litestar to help you to implement the Health Check API pattern

Installation

With pip: 

pip install fast-healthcheck

With poetry: 

poetry add fast-healthcheck

With uv: 

uv add fast-healthcheck

Usage

The easier way to use this package is to use the health function.

Create the health check endpoint dynamically using different conditions. Each condition is a callable, and you can even have dependencies inside of it:

import asyncio
import os
import time
from pathlib import Path
from typing import Any

from dotenv import load_dotenv

from fast_healthchecks.checks import Check
from fast_healthchecks.checks.function import FunctionHealthCheck
from fast_healthchecks.checks.kafka import KafkaHealthCheck
from fast_healthchecks.checks.mongo import MongoHealthCheck
from fast_healthchecks.checks.postgresql.asyncpg import PostgreSQLAsyncPGHealthCheck
from fast_healthchecks.checks.postgresql.psycopg import PostgreSQLPsycopgHealthCheck
from fast_healthchecks.checks.rabbitmq import RabbitMQHealthCheck
from fast_healthchecks.checks.redis import RedisHealthCheck
from fast_healthchecks.checks.url import UrlHealthCheck
from fast_healthchecks.integrations.base import ProbeAsgiResponse

load_dotenv(Path(__file__).parent.parent / ".env")


def sync_dummy_check() -> bool:
    time.sleep(0.1)
    return True


async def async_dummy_check() -> bool:
    await asyncio.sleep(0.1)
    return True


async def async_dummy_check_fail() -> bool:
    msg = "Failed"
    raise Exception(msg) from None  # noqa: TRY002
    await asyncio.sleep(0.1)
    return False


LIVENESS_CHECKS: list[Check] = [
    FunctionHealthCheck(func=sync_dummy_check, name="Sync dummy"),
]

READINESS_CHECKS: list[Check] = [
    KafkaHealthCheck(
        bootstrap_servers=os.environ["KAFKA_BOOTSTRAP_SERVERS"],
        name="Kafka",
    ),
    MongoHealthCheck.from_dsn(os.environ["MONGO_DSN"], name="Mongo"),
    PostgreSQLAsyncPGHealthCheck.from_dsn(os.environ["POSTGRES_DSN"], name="PostgreSQL asyncpg"),
    PostgreSQLPsycopgHealthCheck.from_dsn(os.environ["POSTGRES_DSN"], name="PostgreSQL psycopg"),
    RabbitMQHealthCheck.from_dsn(os.environ["RABBITMQ_DSN"], name="RabbitMQ"),
    RedisHealthCheck.from_dsn(os.environ["REDIS_DSN"], name="Redis"),
    UrlHealthCheck(url="https://httpbin.org/status/200", name="URL 200"),
]

STARTUP_CHECKS: list[Check] = [
    FunctionHealthCheck(func=async_dummy_check, name="Async dummy"),
]

READINESS_CHECKS_SUCCESS: list[Check] = [
    FunctionHealthCheck(func=async_dummy_check, name="Async dummy"),
]
READINESS_CHECKS_FAIL: list[Check] = [
    FunctionHealthCheck(func=async_dummy_check_fail, name="Async dummy fail"),
]


async def custom_handler(response: ProbeAsgiResponse) -> Any:  # noqa: ANN401, RUF029
    """Custom handler for probes."""
    return response.data

from fastapi import FastAPI, status

from examples.probes import (
    LIVENESS_CHECKS,
    READINESS_CHECKS,
    READINESS_CHECKS_FAIL,
    READINESS_CHECKS_SUCCESS,
    STARTUP_CHECKS,
    custom_handler,
)
from fast_healthchecks.integrations.fastapi import HealthcheckRouter, Probe

app_integration = FastAPI()
app_integration.include_router(
    HealthcheckRouter(
        Probe(
            name="liveness",
            checks=LIVENESS_CHECKS,
        ),
        Probe(
            name="readiness",
            checks=READINESS_CHECKS,
        ),
        Probe(
            name="startup",
            checks=STARTUP_CHECKS,
        ),
        debug=True,
        prefix="/health",
    ),
)

app_success = FastAPI()
app_success.include_router(
    HealthcheckRouter(
        Probe(
            name="liveness",
            checks=[],
        ),
        Probe(
            name="readiness",
            checks=READINESS_CHECKS_SUCCESS,
        ),
        Probe(
            name="startup",
            checks=[],
        ),
        debug=True,
        prefix="/health",
    ),
)

app_fail = FastAPI()
app_fail.include_router(
    HealthcheckRouter(
        Probe(
            name="liveness",
            checks=[],
        ),
        Probe(
            name="readiness",
            checks=READINESS_CHECKS_FAIL,
        ),
        Probe(
            name="startup",
            checks=[],
        ),
        debug=True,
        prefix="/health",
    ),
)

app_custom = FastAPI()
app_custom.include_router(
    HealthcheckRouter(
        Probe(
            name="liveness",
            checks=[],
            summary="Check if the application is alive",
        ),
        Probe(
            name="readiness",
            checks=READINESS_CHECKS_SUCCESS,
            summary="Check if the application is ready",
        ),
        Probe(
            name="startup",
            checks=[],
            summary="Check if the application is starting up",
        ),
        success_handler=custom_handler,
        failure_handler=custom_handler,
        success_status=status.HTTP_200_OK,
        failure_status=status.HTTP_503_SERVICE_UNAVAILABLE,
        debug=True,
        prefix="/custom_health",
    ),
)

import os
from http import HTTPStatus

from faststream.asgi import AsgiFastStream
from faststream.kafka import KafkaBroker

from examples.probes import (
    LIVENESS_CHECKS,
    READINESS_CHECKS,
    READINESS_CHECKS_FAIL,
    READINESS_CHECKS_SUCCESS,
    STARTUP_CHECKS,
    custom_handler,
)
from fast_healthchecks.integrations.faststream import Probe, health

broker = KafkaBroker(os.environ["KAFKA_BOOTSTRAP_SERVERS"].split(","))
app_integration = AsgiFastStream(
    broker,
    asgi_routes=[
        *health(
            Probe(name="liveness", checks=LIVENESS_CHECKS),
            Probe(name="readiness", checks=READINESS_CHECKS),
            Probe(name="startup", checks=STARTUP_CHECKS),
            debug=False,
            prefix="/health",
        ),
    ],
)

app_success = AsgiFastStream(
    broker,
    asgi_routes=[
        *health(
            Probe(name="liveness", checks=[]),
            Probe(name="readiness", checks=READINESS_CHECKS_SUCCESS),
            Probe(name="startup", checks=[]),
            debug=False,
            prefix="/health",
        ),
    ],
)

app_fail = AsgiFastStream(
    broker,
    asgi_routes=[
        *health(
            Probe(name="liveness", checks=[]),
            Probe(name="readiness", checks=READINESS_CHECKS_FAIL),
            Probe(name="startup", checks=[]),
            debug=False,
            prefix="/health",
        ),
    ],
)

app_custom = AsgiFastStream(
    broker,
    asgi_routes=[
        *health(
            Probe(
                name="liveness",
                checks=[],
                summary="Check if the application is alive",
            ),
            Probe(
                name="readiness",
                checks=READINESS_CHECKS_SUCCESS,
                summary="Check if the application is ready",
            ),
            Probe(
                name="startup",
                checks=[],
                summary="Check if the application is starting up",
            ),
            success_handler=custom_handler,
            failure_handler=custom_handler,
            success_status=HTTPStatus.OK,
            failure_status=HTTPStatus.SERVICE_UNAVAILABLE,
            debug=True,
            prefix="/custom_health",
        ),
    ],
)

from litestar import Litestar
from litestar.status_codes import HTTP_200_OK, HTTP_503_SERVICE_UNAVAILABLE

from examples.probes import (
    LIVENESS_CHECKS,
    READINESS_CHECKS,
    READINESS_CHECKS_FAIL,
    READINESS_CHECKS_SUCCESS,
    STARTUP_CHECKS,
    custom_handler,
)
from fast_healthchecks.integrations.litestar import Probe, health

app_integration = Litestar(
    route_handlers=[
        *health(
            Probe(name="liveness", checks=LIVENESS_CHECKS),
            Probe(name="readiness", checks=READINESS_CHECKS),
            Probe(name="startup", checks=STARTUP_CHECKS),
            debug=False,
            prefix="/health",
        ),
    ],
)

app_success = Litestar(
    route_handlers=[
        *health(
            Probe(name="liveness", checks=[]),
            Probe(name="readiness", checks=READINESS_CHECKS_SUCCESS),
            Probe(name="startup", checks=[]),
            debug=False,
            prefix="/health",
        ),
    ],
)

app_fail = Litestar(
    route_handlers=[
        *health(
            Probe(name="liveness", checks=[]),
            Probe(name="readiness", checks=READINESS_CHECKS_FAIL),
            Probe(name="startup", checks=[]),
            debug=False,
            prefix="/health",
        ),
    ],
)

app_custom = Litestar(
    route_handlers=[
        *health(
            Probe(
                name="liveness",
                checks=[],
                summary="Check if the application is alive",
            ),
            Probe(
                name="readiness",
                checks=READINESS_CHECKS_SUCCESS,
                summary="Check if the application is ready",
            ),
            Probe(
                name="startup",
                checks=[],
                summary="Check if the application is starting up",
            ),
            success_handler=custom_handler,
            failure_handler=custom_handler,
            success_status=HTTP_200_OK,
            failure_status=HTTP_503_SERVICE_UNAVAILABLE,
            debug=True,
            prefix="/custom_health",
        ),
    ],
)

You can find examples for each framework here: