Spiff Workflow Real Example

nuclear.json

{
    "task_specs": {
        "Start": {
            "class": "SpiffWorkflow.specs.StartTask.StartTask",
            "manual": false,
            "outputs": [
                "general"
            ]
        },
        "general": {
            "class": "SpiffWorkflow.specs.ExclusiveChoice.ExclusiveChoice",
            "name": "general",
            "manual": true,
            "inputs": [
                "Start"
            ],
            "outputs": [
                "workflow_aborted",
                "president"
            ],
            "choice": null,
            "default_task_spec": "workflow_aborted",
            "cond_task_specs": [
                [
                    [
                        "SpiffWorkflow.operators.Equal",
                        [
                            [
                                "Attrib",
                                "confirmation"
                            ],
                            [
                                "value",
                                "yes"
                            ]
                        ]
                    ],
                    "president"
                ]
            ]
        },
        "president": {
            "class": "SpiffWorkflow.specs.ExclusiveChoice.ExclusiveChoice",
            "name": "president",
            "manual": true,
            "inputs": [
                "general"
            ],
            "outputs": [
                "workflow_aborted",
                "nuclear_strike"
            ],
            "choice": null,
            "default_task_spec": "workflow_aborted",
            "cond_task_specs": [
                [
                    [
                        "SpiffWorkflow.operators.Equal",
                        [
                            [
                                "Attrib",
                                "confirmation"
                            ],
                            [
                                "value",
                                "yes"
                            ]
                        ]
                    ],
                    "nuclear_strike"
                ]
            ]
        },
        "nuclear_strike": {
            "class": "SpiffWorkflow.specs.Simple.Simple",
            "name": "nuclear_strike",
            "inputs": [
                "president"
            ],
            "outputs":[
            ]
        },
        "workflow_aborted": {
            "class": "SpiffWorkflow.specs.Cancel.Cancel",
            "name": "workflow_aborted",
            "inputs": [
                "general",
                "president"
            ]
        }
    },
    "description": "",
    "file": null,
    "name": ""
}

nuclear.py

from __future__ import print_function
import os
import json
from SpiffWorkflow import Workflow
from SpiffWorkflow.task import Task
from SpiffWorkflow.serializer.json import JSONSerializer
from SpiffWorkflow.specs import *
from SpiffWorkflow.operators import Equal, Attrib ,Assign
import time


def wait(workflow,nuclear_strike):
    time.sleep(3)
    print('general ok')

def my_nuclear_strike(workflow,nuclear_strike):
    print("sent")

def my_start(workflow,Start):
    general_input=raw_input('general choice:')
    a=workflow.get_tasks_from_spec_name('general')[0]
    a.set_data(confirmation=general_input)

def my_general(workflow,general):
    a=workflow.get_tasks_from_spec_name('general')[0]
    choice=a.get_data('confirmation')
    if choice!='yes':
        print('not send nuclear')
        return
    president_input=raw_input('president choice:')
    a=workflow.get_tasks_from_spec_name('president')[0]
    a.set_data(choose=president_input)


def my_president(workflow,president):
    a=workflow.get_tasks_from_spec_name('president')[0]
    choice=a.get_data('choose')

    if choice!='yes':
        print('not send nuclear')
        return
    
def example(workflow,msg):
    print('perfect')

with open('nuclear.json') as fp:
    workflow_json = fp.read()


serializer = JSONSerializer()
spec = WorkflowSpec.deserialize(serializer, workflow_json)
workflow = Workflow(spec)

Start=workflow.get_task_spec_from_name('Start')
Start.completed_event.connect(my_start)


general=workflow.get_task_spec_from_name('general')
general.completed_event.connect(my_general)

general.ready_event.connect(wait)


president=workflow.get_task_spec_from_name('president')
president.completed_event.connect(my_president)


nuclear_strike=workflow.get_task_spec_from_name('nuclear_strike')
nuclear_strike.completed_event.connect(my_nuclear_strike)

workflow.complete_all(halt_on_manual=False)