Skip to main content

Tiny asynchronous implementation of healthcheck provider and http-server

Project description

aio_tiny_healthcheck

PyPI Build Status Build Status

Tiny asynchronous implementation of healthcheck provider and server

Installation

pip install aio-tiny-healthcheck

Usage

By default, the Checker returns 200 if all checks successfully finish or 500 in opposite case.

Using with aiohttp

from aiohttp import web

from aio_tiny_healthcheck.checker import Checker

def some_sync_check():
    return True

async def some_async_check():
    return False

healthcheck_provider = Checker()
healthcheck_provider.add_check('sync_check_true', some_async_check)
healthcheck_provider.add_check('async_check_false', some_async_check)


app = web.Application()
app.router.add_get('/healthcheck', healthcheck_provider.aiohttp_handler)
web.run_app(app)

Using with Sanic

from sanic import Sanic
from sanic.response import json
from aio_tiny_healthcheck.checker import Checker

app = Sanic()

def some_sync_check():
    return True

async def some_async_check():
    return False

healthcheck_provider = Checker(success_code=201, fail_code=400)
healthcheck_provider.add_check('sync_check_true', some_async_check)
healthcheck_provider.add_check('async_check_false', some_async_check)

@app.route("/healthcheck")
async def test(request):
    hc_response = healthcheck_provider.check_handler()
    return json(hc_response.body, status=hc_response.code)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

Using in concurrent mode

You should want to run healthcheck in background if you already have some blocking operation in your execution flow. So, you can just use built-in server for this.

from aio_tiny_healthcheck.checker import Checker
from aio_tiny_healthcheck.http_server import HttpServer
import asyncio


async def some_long_task():
    await asyncio.sleep(3600)


def some_sync_check():
    return True


async def some_async_check():
    return True


aio_thc = Checker()
hc_server = HttpServer(
    aio_thc,
    path='/health',
    host='localhost',
    port=9090
)

aio_thc.add_check('sync_check_true', some_async_check)
aio_thc.add_check('async_check_false', some_async_check)


async def main():
    # Run healthcheck concurrently
    asyncio.create_task(hc_server.run())

    # Run long task
    await some_long_task()


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Utility for health checking

python -m aio_tiny_healthcheck http://localhost:9192/healthcheck

Useful for running health check without external dependencies like curl.

By default, concurrent server and health checking utility are working with a port and query path http://localhost:8000/healthcheck. So, if you run concurrent server with no using arguments, you can also run the utility with without arguments python -m aio_tiny_healthcheck.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for aio-tiny-healthcheck, version 1.1.2
Filename, size File type Python version Upload date Hashes
Filename, size aio_tiny_healthcheck-1.1.2-py3-none-any.whl (12.2 kB) File type Wheel Python version py3 Upload date Hashes View hashes
Filename, size aio_tiny_healthcheck-1.1.2.tar.gz (8.8 kB) File type Source Python version None Upload date Hashes View hashes

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page