Create an health check endpoint on your REST API following Health Check RFC draft version 4.
In case you have external dependencies, you should check the health of those dependencies.
If you have an external HTTP resource, you can check its health, as in the following sample:
import healthpy.httpx
status, checks = healthpy.httpx.check("petstore", "https://petstore3.swagger.io/api/v3/openapi.json")
Note: httpx module must be installed to perform HTTP health checks.
Alternatively, you can use requests to perform the exact same check:
import healthpy.requests
status, checks = healthpy.requests.check("petstore", "https://petstore3.swagger.io/api/v3/openapi.json")
If you rely on redis, you should check its health.
redis module must be installed to perform Redis health checks.
import healthpy.redis
status, checks = healthpy.redis.check("redis://redis_url", "redis_key")
Once all checks have been performed you should return the result to your client.
If you performed more than one check, you have to compute an aggregated status from all the checks.
import healthpy
status1 = healthpy.pass_status
status2 = healthpy.warn_status
statusN = healthpy.fail_status
status = healthpy.status(status1, status2, statusN)
By default pass status is "pass", warn status is "warn" and fail status is "fail".
It can be tweaked by setting the value of healthpy.*_status as in the following sample:
import healthpy
healthpy.pass_status = "ok"
healthpy.warn_status = "custom"
healthpy.fail_status = "error"
HTTP response body can be retrieved as a dictionary to be returned as JSON.
import healthpy
status = healthpy.pass_status # replace with the aggregated status
checks = {} # replace with the computed checks
body = healthpy.response_body(status, checks=checks)
Checks results are not mandatory in the response.
import healthpy
status = healthpy.pass_status # replace with the aggregated status
body = healthpy.response_body(status)
HTTP response status code can be retrieved as an integer.
import healthpy
status = healthpy.pass_status # replace with the aggregated status
status_code = healthpy.response_status_code(status)
HTTP response status code should be a bit different for Consul health checks.
import healthpy
status = healthpy.pass_status # replace with the aggregated status
status_code = healthpy.consul_response_status_code(status)
An helper function is available to create a starlette endpoint for Consul health check.
from starlette.applications import Starlette
import healthpy
import healthpy.httpx
import healthpy.redis
from healthpy.starlette import add_consul_health_endpoint
app = Starlette()
async def health_check():
# TODO Replace by your own checks.
status_1, checks_1 = healthpy.httpx.check("my external dependency", "http://url_to_check")
status_2, checks_2 = healthpy.redis.check("redis://redis_url", "key_to_check")
return healthpy.status(status_1, status_2), {**checks_1, **checks_2}
# /health endpoint will call the health_check coroutine.
add_consul_health_endpoint(app, health_check)
Note: starlette module must be installed.
An helper function is available to create a Flask-RestX endpoint for health check.
import flask
import flask_restx
import healthpy
import healthpy.httpx
import healthpy.redis
from healthpy.flask_restx import add_health_endpoint
app = flask.Flask(__name__)
api = flask_restx.Api(app)
async def health_check():
# TODO Replace by your own checks.
status_1, checks_1 = healthpy.httpx.check("my external dependency", "http://url_to_check")
status_2, checks_2 = healthpy.redis.check("redis://redis_url", "key_to_check")
return healthpy.status(status_1, status_2), {**checks_1, **checks_2}
# /health endpoint will call the health_check coroutine.
add_health_endpoint(api, health_check)
Note: flask-restx module must be installed.
An helper function is available to create a Flask-RestX endpoint for Consul health check.
import flask
import flask_restx
import healthpy
import healthpy.httpx
import healthpy.redis
from healthpy.flask_restx import add_consul_health_endpoint
app = flask.Flask(__name__)
api = flask_restx.Api(app)
async def health_check():
# TODO Replace by your own checks.
status_1, checks_1 = healthpy.httpx.check("my external dependency", "http://url_to_check")
status_2, checks_2 = healthpy.redis.check("redis://redis_url", "key_to_check")
return healthpy.status(status_1, status_2), {**checks_1, **checks_2}
# /health endpoint will call the health_check coroutine.
add_consul_health_endpoint(api, health_check)
Note: flask-restx module must be installed.
A pytest
fixture can be used to mock the datetime returned in http health check.
from healthpy.testing import mock_http_health_datetime
def test_http(mock_http_health_datetime):
# Time will be returned as "2018-10-11T15:05:05.663979"
pass # Add your test calling healthpy.http.check
- python 3.7+ must be installed
- Use pip to install module:
python -m pip install healthpy