def update_health_status()

in src/redash_stmo/data_sources/health.py [0:0]

• 40 lines of code
• 6 McCabe index (conditional complexity)

def update_health_status():
    for data_source in models.DataSource.query:
        logger.info(u"job=update_health_status state=start ds_id=%s", data_source.id)

        runtime = None
        query_text = data_source.query_runner.noop_query
        ds_id = str(data_source.id)

        custom_query_env_var = "REDASH_CUSTOM_HEALTH_QUERIES_{data_source_id}".format(
            data_source_id=ds_id
        )
        custom_query = os.environ.get(custom_query_env_var, "")
        query_text = custom_query or query_text

        try:
            start_time = time.time()
            test_connection(data_source.query_runner, query_text)
            runtime = time.time() - start_time
        except NotImplementedError:
            logger.info(
                u"Unable to compute health status without test query for %s",
                data_source.name,
            )
            continue
        except Exception:
            logger.warning(
                u"Failed health check for the data source: %s",
                data_source.name,
                exc_info=1,
            )
            statsd_client.incr("update_health_status.error")
            logger.info(
                u"job=update_health_status state=error ds_id=%s runtime=%.2f",
                data_source.id,
                time.time() - start_time,
            )

        status = {
            "status": "FAIL" if runtime is None else "SUCCESS",
            "last_run": start_time,
            "last_run_human": str(parse_human_time(str(start_time))),
            "runtime": runtime,
        }
        store_health_status(ds_id, data_source.name, query_text, status)