def list_jobs()

in python-batch/gke_batch/batch.py [0:0]

• 26 lines of code
• 4 McCabe index (conditional complexity)

  def list_jobs(self):
    namespace = get_setting("namespace", settings)
    try:
      api_response = self.core_v1.list_namespaced_job(namespace)
      for item in api_response.items:
        succeeded = item.status.succeeded
        failed = item.status.failed
        completed = item.status.completed_indexes
        print(f"Name: {item.metadata.labels['app']}\tSucceeded: {succeeded}\tFailed: {failed}\tCompleted Index: {completed}", file=sys.stderr)
        # pprint(item)
    except ApiException as e:
      print("Exception when calling BatchV1Api->list_namespaced_job: %s\n" % e)
    headers = ["Job Name", "Succeeded", "Failed", "Completed Index"]
    rows = []
    for item in api_response.items:
      succeeded = item.status.succeeded
      failed = item.status.failed
      completed = item.status.completed_indexes
      rows.append(
        [
          colored(item.metadata.labels['job-name'], 'blue'),
          colored(succeeded, 'green'),
          colored(failed, 'red'),
          colored(completed, 'yellow'),
        ]
      )

    print(tabulate(rows, headers, tablefmt="grid"))