storage-service/storage_service/utils/healthchecks.py

22 lines
572 B
Python

#!/usr/bin/env python3
from utils import storage
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 register_health_checks(app: Flask):
health = HealthCheck()
health.add_check(health_database_status)
app.add_url_rule("/healthz", "healthcheck", view_func=lambda: health.run())