storage-service/storage_service/utils/healthchecks.py

#!/usr/bin/env python3
from utils import storage, magic_amqp
from healthcheck import HealthCheck
from flask import Flask


def health_database_status():
    is_database_working = True
    output = 'storage is ok'
    try:
        storage.connection.list_buckets()
    except Exception as e:
        output = str(e)
        is_database_working = False
    return is_database_working, output


def amqp_connection_status():
    if magic_amqp.is_healthy():
        result = True
        text = "amqp connection is ok"
    else:
        result = False
        text = "amqp connection is unhealthy"

    return result, text


def register_health_checks(app: Flask):
    health = HealthCheck()
    health.add_check(health_database_status)
    health.add_check(amqp_connection_status)
    app.add_url_rule("/healthz", "healthcheck", view_func=lambda: health.run())