Set workflow state or retrieve a list of transitions required to get from one state to another

workflow_tools.py

#@memoize
def get_wf_transitions_for(workflow, from_state, to_state):
    exit_state_maps = {}
    for state in workflow.states.objectValues():
        for t in state.getTransitions():
            exit_state_maps.setdefault(t, [])
            exit_state_maps[t].append(state.getId())
            
    transition_maps = {}
    for transition in workflow.transitions.objectValues():
        value = (transition.getId(), exit_state_maps.get(transition.getId(), []))
        if transition.new_state_id not in transition_maps:
            transition_maps[transition.new_state_id] = [value]
        else:
            transition_maps[transition.new_state_id].append(value)
    
    if to_state not in transition_maps:
        # impossible to reach via this workflow
        return None
    
    # work backwards from our end state
    def find_path(maps, path, current_state, start_state):
        for transition, from_states in maps[current_state]:
            if transition in path:
                # Don't go in a circle
                continue
            
            path.insert(0, transition)
            if start_state in from_states:
                return path
                
            for state in from_states:
                find_path(maps, path, state, start_state)
                if path:
                    return path
        return []
    
    return find_path(transition_maps, [], to_state, from_state)


def set_workflow_state(instance, state):
    wftool = getToolByName(instance, 'portal_workflow')
    for wf in wftool.getWorkflowsFor(instance):
        
        status = wftool.getStatusOf(wf.getId(), instance)
        if not status or not status.get('review_state'):
            continue
        
        if status['review_state']==state:
            return
        
        transitions = get_wf_transitions_for(wf, status['review_state'], state)
        
        if not transitions:
            continue
        
        for transition in transitions:
            wftool.doActionFor(instance, transition)
        
        break
        
    if wftool.getInfoFor(instance, 'review_state') != state:
        raise RuntimeError("Could not find workflow to set state to %s on %r" % (state, instance))