kludex / fastapi-health

Implement the Health Check API pattern on your FastAPI application! :rocket:

FastAPI Health 🚑️

The goal of this package is to help you to implement the Health Check API pattern.

Installation

pip install fastapi-health

Quick Start

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

from fastapi import FastAPI, Depends
from fastapi_health import health


def get_session():
    return True


def is_database_online(session: bool = Depends(get_session)):
    return session


app = FastAPI()
app.add_api_route("/health", health([is_database_online]))

Advanced Usage

The health() method receives the following parameters:

• conditions: A list of callables that represents the conditions of your API, it can return either bool or a dict.
• success_handler: An optional callable which receives the conditions results and returns a dictionary that will be the content response of a successful health call.
• failure_handler: An optional callable analogous to success_handler for failure scenarios.
• success_status: An integer that overwrites the default status (200) in case of success.
• failure_status: An integer that overwrites the default status (503) in case of failure.

It's important to notice that you can have a peculiar behavior in case of hybrid return statements (bool and dict) on the conditions. For example:

from fastapi import FastAPI
from fastapi_health import health


def pass_condition():
    return {"database": "online"}


def sick_condition():
    return False


app = FastAPI()
app.add_api_route("/health", health([pass_condition, sick_condition]))

This will generate a response composed by the status being 503 (default failure_status), because sick_condition returns False, and the JSON body {"database": "online"}. It's not wrong, or a bug. It's meant to be like this.

License

This project is licensed under the terms of the MIT license.

fastapi-health's Issues

Should implement RFC draft for health check

The health check module should implement the RFC draft for health check.

The latest official version of the draft can be found here: https://datatracker.ietf.org/doc/html/draft-inadarei-api-health-check-06

To participate in the development of the draft: https://github.com/inadarei/rfc-healthcheck/

For an example of a Python library that implements the Health Check API: https://github.com/Colin-b/healthpy

functools.partial not supported

Using a functools.partial() object in health() results in an error:

  File "/usr/local/lib/python3.10/site-packages/fastapi_health/route.py", line 40, in health
    f"{condition.__name__}",
AttributeError: 'functools.partial' object has no attribute '__name__'

you can use condition.func.__name__ instead.

FastAPI-health library

Hi there,
So how do we use fastapi_health import health library for the checking the endpoints for different files.
so I have 5 files in folder and all 5 of them have different endpoints.
I add new file called healthcheck.py but wondering how can I use fastapi health library to check all endpoints in healthcheck.py file?

5 files has different URL and different app=APIRoutuer, and POST call and GET call.

Please suggest me any idea of implementing healthcheck for my endpoints!

doc: success_output/failure_output args don't exist, should be success/failure_handler...

The README states the arg names for customizing the responses are success_output and failure_output... which is wrong, its the _handler naming instead

RFC draft for health check not supporting APIRouter

I'm using inreleased version of this package

app.add_api_route("/internal/health",
HealthEndpoint(conditions=[live_condition]),
)
app.add_api_route("/internal/ready",
HealthEndpoint(conditions=[ready_condition]),
)
works

r = APIRouter()
r.add_api_route("/internal/health",
HealthEndpoint(conditions=[live_condition]),
)
r.add_api_route("/internal/ready",
HealthEndpoint(conditions=[ready_condition]),
)
app.include_router(r)
fails because it returns ready_condition for both endpoints

Type annotation for `health` doesn't support async methods

from fastapi import Depends, FastAPI
from fastapi_health import health

def get_session():
    return True

def is_database_online(session: bool = Depends(get_session)):
    return session

async def async_condition() -> bool:
    return True

app = FastAPI()
app.add_api_route("/health", health([is_database_online, async_condition]))

The above code fails MyPy check due to the anntoation on health not supporting async functions.

main.py:16: error: List item 1 has incompatible type "Callable[[], Coroutine[Any, Any, bool]]"; expected "Callable[..., dict[str, Any] | bool]"  [list-item]
Found 1 error in 1 file (checked 1 source file)

As outlined in #9, the library does support using async functions, so I think the annotation can be updated to add support.

health return type not compatible `FastAPI.add_api_route`

The FastAPI method add_api_route takes in Callable[..., Coroutine[Any, Any, Response]] but the health method for this library returns Callable[..., Awaitable[JSONResponse]] which raises an error in MyPy.

# test.py
from fastapi import FastAPI
from fastapi_health import health

def is_database_online() -> bool:
    return True

app = FastAPI()
app.add_api_route(
    "/health",
    health([is_database_online]),
)

test.py:12: error: Argument 2 to "add_api_route" of "FastAPI" has incompatible type "Callable[..., Awaitable[JSONResponse]]"; expected "Callable[..., Coroutine[Any, Any, Response]]"  [arg-type]
Found 1 error in 1 file (checked 1 source file)

Changing the method signature to Callable[..., Coroutine[Any, Any, JSONResponse]] would fix these warnings.

MyPy version: 0.991
FastAPI version: 0.89.1

Please add py.typed marker

Using fastapi-health, mypy reports "module is installed but missing library stubs or py.typed marker".

As fastapi is nicely typed, you can easily add a py.typed marker. It's enough to add an empty file called py.typed in the fastapi_health package directory, and to include it in the distribution.

I would provide a MR, but I have no idea how to do include the file in a project using pyproject.toml file. If there was a setup.cfg file, it would be enough to add a stanza such as:

[options.package_data]
fastapi_health = py.typed

Disallow Pydantic v2 until support is added

Currently, this library does not support Pydantic v2 however, there is no requirement laid out in the dependencies. Since the latest version of FastAPI (0.100.1 at the time of writing) still supports Pydantic v1, I think Pydantic v2 support can be disabled until explicitly supported.

Allow `HEAD` request

Hello,

I was wondering if this library can, somehow, allow HEAD requests. I can see that FastAPI, for the moment, does not automatically support HEAD method for GET routes (see: tiangolo/fastapi#1773).

Thanks in advance.

If a condition fails, the health endpoint crashes with a 500

If a condition function throws an exception, the health endpoint fails with a 500. I don't think this is the right thing to do for a function supposed to check for something that might be in failing state.

As a consequence, the way I'm using fastapi-health, in an app using Postgres, RabbitMQ, MinIO, looks like:

health_endpoint = health([database_check, rabbitmq_check, storage_check])
app.add_api_route("/health", health_endpoint, include_in_schema=False)

async def database_check() -> bool:
    try:
        async with db.pool.connection(timeout=5) as conn:
            await conn.execute("select 1")
        return True
    except Exception as e:
        logger.error("error checking database: %s - %s", type(e).__name__, e)
        return False

async def rabbitmq_check() -> bool:
    try:
        await mq.mq.check_health()
        return True
    except Exception as e:
        logger.error("error checking rabbitmq: %s - %s", type(e).__name__, e)
        return False

async def storage_check() -> bool:
    try:
        await FileUploader.check_health(minio_url)
        return True
    except Exception as e:
        logger.error("error checking storage: %s - %s", type(e).__name__, e)
        return False

I think that the health endpoint should be just tolerant of failure and consider an error thrown by a condition function as a failure. I should be able to write the above as:

async def database_check() -> bool:
    async with db.pool.connection(timeout=5) as conn:
        await conn.execute("select 1")
    return True

async def rabbitmq_check() -> bool:
    await mq.mq.check_health()
    return True

async def storage_check() -> bool:
    await FileUploader.check_health(minio_url)
    return True

For me the return True is redundant too, but that's probably a different matter.

Thoughts?

How to get a response body in fastapi-health as return

How do we response body and not just response headers in return as the route.py body only returns 200 and 503 response status codes only

async support

it would be great if there was async support here, for example with something like this:

app.add_api_route("/health", health([sync_methods], async_health([async_methods])))

fastapi-health would take care of awaiting the async_health methods etc. My use case is that in my async-only fastapi app I have to implement async-in-sync wrappers in order to use fastapi-health as it is today.

How to return a failure + details about why?

I don't see how I can get this to return a 503 (fail) + a custom detail as to why a particular check failed.

Seems that if I return any dict from a check function, that is treated as success. They only way to have a check "fail" is to return None or False.... and in those cases I can't return a dict of details....

anyway to acheive this?