Function migrate_workflow

SpiffWorkflow/bpmn/util/diff.py:213–245

Update the spec for workflow. Args: diff (`WorkflowDiff`): the diff of this workflow and spec workflow (`BpmnWorkflow` or `BpmnSubWorkflow`): the workflow spec (`BpmnProcessSpec`): the new spec Keyword Args: reset_mask (`TaskState`): reset and repredict task

(diff, workflow, spec, reset_mask=None)

    return [t for t in tasks if task_filter.matches(t)]

def migrate_workflow(diff, workflow, spec, reset_mask=None):
    """Update the spec for workflow.

    Args:
        diff (`WorkflowDiff`): the diff of this workflow and spec
        workflow (`BpmnWorkflow` or `BpmnSubWorkflow`): the workflow
        spec (`BpmnProcessSpec`): the new spec

    Keyword Args:
        reset_mask (`TaskState`): reset and repredict tasks in this state

    Returns
        a list of deleted `Task`

    The default rest_mask is TaskState.READY|TaskState.WAITING but can be overridden.
    """
    workflow.spec = spec
    for task in workflow.tasks.values():
        if diff.alignment.get(task) is not None:
            task.task_spec = diff.alignment.get(task)

    default_mask = TaskState.READY|TaskState.WAITING
    removed_tasks = []
    for task in list(workflow.get_tasks(state=reset_mask or default_mask, skip_subprocesses=True)):
        # In some cases, completed tasks with ready or waiting children could get removed
        # (for example, in cycle timer).  If a task has already been removed from the tree, ignore it.
        if task.id in workflow.tasks:
            removed_tasks.extend(task.reset_branch(None))
    
    if workflow.last_task not in workflow.tasks:
        workflow.last_task = None

    return removed_tasks