Tutorial - Non-BPMN

Introduction

In this chapter we are going to use Spiff Workflow to solve a real-world problem: We will create a workflow for triggering a nuclear strike.

We are assuming that you are familiar with the Fundamental SpiffWorkflow Concepts.

Assume you want to send the rockets, but only after both the president and a general have signed off on it.

There are two different ways of defining a workflow: Either by deserializing (from XML or JSON), or using Python.

Creating the workflow specification (using Python)

As a first step, we are going to create a simple workflow in code. In Python, the workflow is defined as follows:

from SpiffWorkflow.specs.WorkflowSpec import WorkflowSpec
from SpiffWorkflow.specs.ExclusiveChoice import ExclusiveChoice
from SpiffWorkflow.specs.Simple import Simple
from SpiffWorkflow.specs.Cancel import Cancel
from SpiffWorkflow.operators import Equal, Attrib

def my_nuclear_strike(msg):
    print("Launched:", msg)

class NuclearStrikeWorkflowSpec(WorkflowSpec):
    def __init__(self):
        WorkflowSpec.__init__(self)

        # The first step of our workflow is to let the general confirm
        # the nuclear strike.
        general_choice = ExclusiveChoice(self, 'general')
        self.start.connect(general_choice)

        # The default choice of the general is to abort.
        cancel = Cancel(self, 'workflow_aborted')
        general_choice.connect(cancel)

        # Otherwise, we will ask the president to confirm.
        president_choice = ExclusiveChoice(self, 'president')
        cond = Equal(Attrib('confirmation'), 'yes')
        general_choice.connect_if(cond, president_choice)

        # The default choice of the president is to abort.
        president_choice.connect(cancel)

        # Otherwise, we will perform the nuclear strike.
        strike = Simple(self, 'nuclear_strike')
        president_choice.connect_if(cond, strike)

        # Now we connect our Python function to the Task named 'nuclear_strike'
        strike.completed_event.connect(my_nuclear_strike)

        # As soon as all tasks are either "completed" or  "aborted", the
        # workflow implicitely ends.

Hopefully the code is self explaining. Using Python to write a workflow can quickly become tedious. It is usually a better idea to use another format.

Creating a workflow specification (using JSON)

Once you have completed the serializer as shown above, you can write the specification in JSON.

Here is an example that is doing exactly the same as the Python WorkflowSpec above:

{
    "task_specs": {
        "Start": {
            "class": "SpiffWorkflow.specs.StartTask.StartTask",
	        "id" : 1,
            "manual": false,
            "outputs": [
                "general"
            ]
        },
        "general": {
            "class": "SpiffWorkflow.specs.ExclusiveChoice.ExclusiveChoice",
            "name": "general",
	        "id" : 2,
            "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",
	        "id" : 3,
            "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": {
	        "id" : 4,
            "class": "SpiffWorkflow.specs.Simple.Simple",
            "name": "nuclear_strike",
            "inputs": [
                "president"
            ]
        },
        "workflow_aborted": {
	    "id" : 5,
            "class": "SpiffWorkflow.specs.Cancel.Cancel",
            "name": "workflow_aborted",
            "inputs": [
                "general",
                "president"
            ]
        }
    },
    "description": "",
    "file": null,
    "name": ""
}

Creating a workflow out of the specification

Now it is time to get started and actually create and execute a workflow according to the specification.

Since we included manual tasks in the specification, you will want to implement a user interface in practice, but we are just going to assume that all tasks are automatic for this tutorial. Note that the manual flag has no effect on the control flow; it is just a flag that a user interface may use to identify tasks that require a user input.

from SpiffWorkflow.workflow import Workflow
from SpiffWorkflow.specs.WorkflowSpec import WorkflowSpec
from SpiffWorkflow.serializer.json import JSONSerializer

# Load from JSON
with open('nuclear.json') as fp:
    workflow_json = fp.read()
serializer = JSONSerializer()
spec = WorkflowSpec.deserialize(serializer, workflow_json)

# Alternatively, create an instance of the Python based specification.
#from nuclear import NuclearStrikeWorkflowSpec
#spec = NuclearStrikeWorkflowSpec()

# Create the workflow.
workflow = Workflow(spec)

# Execute until all tasks are done or require manual intervention.
# For the sake of this tutorial, we ignore the "manual" flag on the
# tasks. In practice, you probably don't want to do that.
workflow.run_all(halt_on_manual=False)

# Alternatively, this is what a UI would do for a manual task.
#workflow.complete_task_from_id(...)

SpiffWorkflow.Workflow.complete_all() completes all tasks in accordance to the specification, until no further tasks are READY for being executed. Note that this does not mean that the workflow is completed after calling SpiffWorkflow.Workflow.complete_all(), since some tasks may be WAITING, or may be blocked by another WAITING task, for example.

Serializing a workflow

If you want to store a SpiffWorkflow.specs.WorkflowSpec, you can use SpiffWorkflow.specs.WorkflowSpec.serialize():

import json
from SpiffWorkflow.serializer.json import JSONSerializer
from nuclear import NuclearStrikeWorkflowSpec

serializer = JSONSerializer()
spec = NuclearStrikeWorkflowSpec()
data = spec.serialize(serializer)

# This next line is unnecessary in practice; it just makes the JSON pretty.
pretty = json.dumps(json.loads(data), indent=4, separators=(',', ': '))

open('workflow-spec.json', 'w').write(pretty)

If you want to store a SpiffWorkflow.Workflow, use use SpiffWorkflow.Workflow.serialize():

import json
from SpiffWorkflow.workflow import Workflow
from SpiffWorkflow.serializer.json import JSONSerializer
from nuclear import NuclearStrikeWorkflowSpec

serializer = JSONSerializer()
spec = NuclearStrikeWorkflowSpec()
workflow = Workflow(spec)
data = workflow.serialize(serializer)

# This next line is unnecessary in practice; it just makes the JSON pretty.
pretty = json.dumps(json.loads(data), indent=4, separators=(',', ': '))

open('workflow.json', 'w').write(pretty)

Deserializing a workflow

The following example shows how to restore a SpiffWorkflow.specs.WorkflowSpec using SpiffWorkflow.specs.WorkflowSpec.serialize().

from SpiffWorkflow.specs.WorkflowSpec import WorkflowSpec
from SpiffWorkflow.serializer.json import JSONSerializer

serializer = JSONSerializer()
with open('workflow-spec.json') as fp:
    workflow_json = fp.read()
spec = WorkflowSpec.deserialize(serializer, workflow_json)

To restore a SpiffWorkflow.Workflow, use SpiffWorkflow.Workflow.serialize() instead:

from SpiffWorkflow.workflow import Workflow
from SpiffWorkflow.serializer.json import JSONSerializer

serializer = JSONSerializer()
with open('workflow.json') as fp:
    workflow_json = fp.read()
workflow = Workflow.deserialize(serializer, workflow_json)

Where to go from here?

This first tutorial actually has a problem: If you want to save the workflow, SpiffWorkflow won’t be able to re-connect the signals because it can not save the reference to your code.

So after deserializing the workflow, you will need to re-connect the signals yourself.

If you would rather have it such that SpiffWorkflow handles this for you, you need to create a custom task and tell SpiffWorkflow how to serialize and deserialize it. The next tutorial shows how this is done.