def _check_job_running()

in exec/src/klio_exec/commands/stop.py [0:0]

• 28 lines of code
• 5 McCabe index (conditional complexity)

def _check_job_running(config):
    request = (
        _client.projects()
        .locations()
        .jobs()
        .list(
            projectId=config.pipeline_options.project,
            location=config.pipeline_options.region,
            filter="ACTIVE",
        )
    )

    try:
        response = request.execute()
    except Exception as e:
        logging.warning(
            "Could not find running job '{}' in project '{}': {}".format(
                config.job_name, config.pipeline_options.project, e
            )
        )
        logging.warning(
            "Continuing to attempt deploying '{}'".format(config.job_name)
        )
        return

    job_results = response.get("jobs", [])
    if job_results:
        for result in job_results:
            if result["name"] == config.job_name:
                return result