Class TaskIterator

SpiffWorkflow/util/task.py:179–267

Default task iteration class.

        

class TaskIterator:
    """Default task iteration class."""

    def __init__(self, task, end_at_spec=None, max_depth=1000, depth_first=True, task_filter=None, **kwargs):
        """Iterate over the task tree and return the tasks matching the filter parameters.

        Args:
            task (`Task`):    the task to start from

        Keyword Args:
            end_at (str):stop when a task spec with this name is reached
            max_depth (int): stop when this depth is reached
            depth_first (bool): return results in depth first order
            task_filter (`TaskFilter`): return only tasks matching this filter
        
        Notes:
            Keyword args not used by this class will be passed into `TaskFilter` if no `task_filter` is provided.
            This is for convenience (filter values can be used directly from `Workflow.get_tasks`) as well as
            backwards compatilibity for queries about `TaskState`.
        """
        self.task_filter = task_filter or TaskFilter(**kwargs)
        self.end_at_spec = end_at_spec
        self.max_depth = max_depth
        self.depth_first = depth_first

        self.task_list = [task]
        self.depth = 0
        # Figure out which states need to be traversed.
        # Predicted tasks can follow definite tasks but not vice versa; definite tasks can follow finished tasks but not vice versa
        # The dream is for a child task to always have a lower task state than its parent; currently we have parents that wait for
        # their children
        if self.task_filter.state & TaskState.PREDICTED_MASK:
            self.min_state = TaskState.MAYBE
        elif self.task_filter.state & TaskState.DEFINITE_MASK:
            self.min_state = TaskState.FUTURE
        else:
            self.min_state = TaskState.COMPLETED

    def __iter__(self):
        return self

    def __next__(self):
        task = self._next()
        while not self.task_filter.matches(task):
            task = self._next()
        return task

    def _next(self):

        if not self.task_list:
            raise StopIteration()

        task = self.task_list.pop(0)
        if task._children and \
            task.state >= self.min_state and \
            self.depth < self.max_depth and \
            task.task_spec.name != self.end_at_spec: