quay/data/model/health.py

import logging

from data.database import TeamRole, validate_database_url

logger = logging.getLogger(__name__)

def check_health(app_config):
  # Attempt to connect to the database first. If the DB is not responding,
  # using the validate_database_url will timeout quickly, as opposed to
  # making a normal connect which will just hang (thus breaking the health
  # check).
  try:
    validate_database_url(app_config['DB_URI'], {}, connect_timeout=3)
  except Exception as ex:
    return (False, 'Could not connect to the database: %s' % ex.message)

  # We will connect to the db, check that it contains some team role kinds
  try:
    return (bool(list(TeamRole.select().limit(1))), 'Could not connect to the database')
  except Exception as ex:
    return (False, 'Could not connect to the database: %s' % ex.message)