Tiny asynchronous implementation of healthcheck provider and http-server
Project description
aio_tiny_healthcheck
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.
Source Distribution
Built Distribution
Hashes for aio_tiny_healthcheck-1.1.2.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 6cdd90bf24b6a0fd30ba4884256046be5bea914c9b0dc6dca2ba1007beccf139 |
|
MD5 | 02b55c06a5e80826012f19b9313d5638 |
|
BLAKE2b-256 | 01a6b58735ef01f8a30f52ce5146c9b185c08a52706b373c8074dcb601053d97 |
Hashes for aio_tiny_healthcheck-1.1.2-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | ef5fbff45611fbcae9bf44900a37f6a1ca98e087e94fcaaf9a5b689f02a94ecf |
|
MD5 | d14f5f0f8e1f8a0f9f3d8c3ccce950fa |
|
BLAKE2b-256 | 73e093a677cef10c36e6e11b0926af7011c7af30f4baa4664b1d9829999e64d6 |