def get_next_step()

in experiments/task_dispatch_service/{{destination_path}}/{{component_name}}/src/workflow/workflow.py [0:0]

• 10 lines of code
• 8 McCabe index (conditional complexity)

  def get_next_step(self, task):
    current_complete_step = None

    # Find the current complete step in order to find the next step.
    for step in self.workflow["steps"]:
      # Found complete step, hence returning the next step object.
      if current_complete_step:
        return step

      # If not found, keep checking if the step names are the same and
      # the status is "complete".
      elif task.step == step["name"] and task.status == str(
          TaskStatus.COMPLETE.value):
        current_complete_step = step
        print(f"current_complete_step = {current_complete_step}")

    # If no current complete step found, either the next step is not ready, or
    # there's no next step.
    return None