项目作者: Colin-b

项目描述 :
Health Check for HTTP APIs (RFC implementation)
高级语言: Python
项目地址: git://github.com/Colin-b/healthpy.git
创建时间: 2019-11-28T22:30:37Z
项目社区:https://github.com/Colin-b/healthpy

开源协议:MIT License

下载


Health Check for HTTP APIs


pypi version
Build status
Coverage
Code style: black
Number of tests
Number of downloads

Create an health check endpoint on your REST API following Health Check RFC draft version 4.

Perform checks

In case you have external dependencies, you should check the health of those dependencies.

HTTP

If you have an external HTTP resource, you can check its health, as in the following sample:

  1. import healthpy.httpx
  2. 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:

  1. import healthpy.requests
  2. status, checks = healthpy.requests.check("petstore", "https://petstore3.swagger.io/api/v3/openapi.json")

Redis

If you rely on redis, you should check its health.

redis module must be installed to perform Redis health checks.

  1. import healthpy.redis
  2. status, checks = healthpy.redis.check("redis://redis_url", "redis_key")

Return result

Once all checks have been performed you should return the result to your client.

Compute status from multiple statuses

If you performed more than one check, you have to compute an aggregated status from all the checks.

  1. import healthpy
  2. status1 = healthpy.pass_status
  3. status2 = healthpy.warn_status
  4. statusN = healthpy.fail_status
  5. status = healthpy.status(status1, status2, statusN)

Using custom status

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:

  1. import healthpy
  2. healthpy.pass_status = "ok"
  3. healthpy.warn_status = "custom"
  4. healthpy.fail_status = "error"

HTTP response body

HTTP response body can be retrieved as a dictionary to be returned as JSON.

  1. import healthpy
  2. status = healthpy.pass_status # replace with the aggregated status
  3. checks = {} # replace with the computed checks
  4. body = healthpy.response_body(status, checks=checks)

Checks results are not mandatory in the response.

  1. import healthpy
  2. status = healthpy.pass_status # replace with the aggregated status
  3. body = healthpy.response_body(status)

HTTP response status code

HTTP response status code can be retrieved as an integer.

  1. import healthpy
  2. status = healthpy.pass_status # replace with the aggregated status
  3. status_code = healthpy.response_status_code(status)

Consul

HTTP response status code should be a bit different for Consul health checks.

  1. import healthpy
  2. status = healthpy.pass_status # replace with the aggregated status
  3. status_code = healthpy.consul_response_status_code(status)

Endpoint

Starlette

An helper function is available to create a starlette endpoint for Consul health check.

  1. from starlette.applications import Starlette
  2. import healthpy
  3. import healthpy.httpx
  4. import healthpy.redis
  5. from healthpy.starlette import add_consul_health_endpoint
  6. app = Starlette()
  7. async def health_check():
  8. # TODO Replace by your own checks.
  9. status_1, checks_1 = healthpy.httpx.check("my external dependency", "http://url_to_check")
  10. status_2, checks_2 = healthpy.redis.check("redis://redis_url", "key_to_check")
  11. return healthpy.status(status_1, status_2), {**checks_1, **checks_2}
  12. # /health endpoint will call the health_check coroutine.
  13. add_consul_health_endpoint(app, health_check)

Note: starlette module must be installed.

Flask-RestX

An helper function is available to create a Flask-RestX endpoint for health check.

  1. import flask
  2. import flask_restx
  3. import healthpy
  4. import healthpy.httpx
  5. import healthpy.redis
  6. from healthpy.flask_restx import add_health_endpoint
  7. app = flask.Flask(__name__)
  8. api = flask_restx.Api(app)
  9. async def health_check():
  10. # TODO Replace by your own checks.
  11. status_1, checks_1 = healthpy.httpx.check("my external dependency", "http://url_to_check")
  12. status_2, checks_2 = healthpy.redis.check("redis://redis_url", "key_to_check")
  13. return healthpy.status(status_1, status_2), {**checks_1, **checks_2}
  14. # /health endpoint will call the health_check coroutine.
  15. add_health_endpoint(api, health_check)

Note: flask-restx module must be installed.

Consul Service Health check

An helper function is available to create a Flask-RestX endpoint for Consul health check.

  1. import flask
  2. import flask_restx
  3. import healthpy
  4. import healthpy.httpx
  5. import healthpy.redis
  6. from healthpy.flask_restx import add_consul_health_endpoint
  7. app = flask.Flask(__name__)
  8. api = flask_restx.Api(app)
  9. async def health_check():
  10. # TODO Replace by your own checks.
  11. status_1, checks_1 = healthpy.httpx.check("my external dependency", "http://url_to_check")
  12. status_2, checks_2 = healthpy.redis.check("redis://redis_url", "key_to_check")
  13. return healthpy.status(status_1, status_2), {**checks_1, **checks_2}
  14. # /health endpoint will call the health_check coroutine.
  15. add_consul_health_endpoint(api, health_check)

Note: flask-restx module must be installed.

Testing

A pytest fixture can be used to mock the datetime returned in http health check.

  1. from healthpy.testing import mock_http_health_datetime
  2. def test_http(mock_http_health_datetime):
  3. # Time will be returned as "2018-10-11T15:05:05.663979"
  4. pass # Add your test calling healthpy.http.check

How to install

  1. python 3.7+ must be installed
  2. Use pip to install module:
    1. python -m pip install healthpy