Skip to content

Instantly share code, notes, and snippets.

@iraycd
Created December 12, 2018 15:12
Show Gist options
  • Save iraycd/b80951a54a14c09bb1d73f8f79b496b2 to your computer and use it in GitHub Desktop.
Save iraycd/b80951a54a14c09bb1d73f8f79b496b2 to your computer and use it in GitHub Desktop.
Spiff Workflow Real Example
{
"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": ""
}
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment