def call_tasks()

in gce_rescue/tasks/actions.py [0:0]

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

def call_tasks(vm: Instance, action: str) -> None:
  """ Loop tasks dict and execute """
  tasks = _list_tasks(vm = vm, action = action)
  async_backup_thread = None
  if action == 'set_rescue_mode':
    if get_config('skip-snapshot'):
      _logger.info(f'Skipping snapshot backup.')
    else:
      take_snapshot(vm)
      async_backup_thread = True
  total_tasks = len(tasks)

  tracker = Tracker(total_tasks)
  tracker.start()

  for task in tasks:
    execute = task['name']
    args = task['args'][0]

    execute(**args)
    tracker.advance(step = 1)

  if async_backup_thread:
    _logger.info(f'Waiting for async backup to finish')
    take_snapshot(vm, join_snapshot=True)
    _logger.info('done.')
  tracker.finish()