Method _sync_children

SpiffWorkflow/task.py:234–272

Syncs the task's children with the given list of task specs. - Add one child for each given `TaskSpec`, unless that child already exists. - Remove all children for which there is no spec in the given list, unless it is a "triggered" task. Notes: It is an err

(self, task_specs, state=TaskState.MAYBE)

        return task

    def _sync_children(self, task_specs, state=TaskState.MAYBE):
        """Syncs the task's children with the given list of task specs. 
    
        - Add one child for each given `TaskSpec`, unless that child already exists.
        - Remove all children for which there is no spec in the given list, unless it is a "triggered" task.

        Notes:
           It is an error if the task has a non-predicted child that is not given in the TaskSpecs.

        Args:
            task_spec (list(`TaskSpec`)): the list of task specs that may become children
            state (`TaskState`): the state to assign
        """
        if task_specs is None:
            raise ValueError('"task_specs" argument is None')
        new_children = task_specs[:]

        # Create a list of all children that are no longer needed.
        unneeded_children = []
        for child in self.children:
            if child.triggered:
                # Triggered tasks are never removed.
                pass
            elif child.task_spec in new_children:
                # If the task already exists, remove it from to-be-added and update its state
                new_children.remove(child.task_spec)
                if child.has_state(TaskState.NOT_FINISHED_MASK):
                    child._set_state(state)
            else:
                if child.has_state(TaskState.DEFINITE_MASK):
                    # Definite tasks must not be removed, so they HAVE to be in the given task spec list.
                    raise WorkflowException(f'removal of non-predicted child {child}', task_spec=self.task_spec)
                unneeded_children.append(child)

        # Update children accordingly
        for child in unneeded_children:
            self.workflow._remove_task(child.id)
        for task_spec in new_children:
            self._add_child(task_spec, state)

    def _child_added_notify(self, child):
        """Called by another task to let us know that a child was added."""