Implementing Custom Tasks

Introduction

In this second tutorial, we are going to implement our own task, and use serialization and deserialization to store and restore it.

If you haven’t already, you should complete the first Tutorial - Non-BPMN. We are also assuming that you are familiar with the Fundamental SpiffWorkflow Concepts.

Implementing the custom task

The first step is to create a SpiffWorkflow.specs.TaskSpec that fires the rocket:

from SpiffWorkflow.specs import Simple

class NuclearStrike(Simple):
    def _on_complete_hook(self, my_task):
        print("Rocket sent!")

Save this file as strike.py.

Now, before we are ready to define the workflow using XML or JSON, we will also have extend the serializer to let SpiffWorkflow know how to represent your NuclearStrike first.

Preparing a serializer

Before we can use JSON to specify a workflow, we first need to teach SpiffWorkflow what our custom NuclearChoice looks like in JSON. We do this by extending the SpiffWorkflow.serializer.json.JSONSerializer.

from SpiffWorkflow.serializer.json import JSONSerializer
from strike import NuclearStrike

class NuclearSerializer(JSONSerializer):
    def serialize_nuclear_strike(self, task_spec):
        return self.serialize_task_spec(task_spec)

    def deserialize_nuclear_strike(self, wf_spec, s_state):
        spec = NuclearStrike(wf_spec, s_state['name'])
        self.deserialize_task_spec(wf_spec, s_state, spec=spec)
        return spec

We save the serializer as serializer.py. We also need to update strike.py as follows:

We also implement the deserializer:

from SpiffWorkflow.specs.Simple import Simple

class NuclearStrike(Simple):
    def _on_complete_hook(self, my_task):
        print((self.my_variable, "sent!"))

    def serialize(self, serializer):
        return serializer.serialize_nuclear_strike(self)

    @classmethod
    def deserialize(self, serializer, wf_spec, s_state):
        return serializer.deserialize_nuclear_strike(wf_spec, s_state)

That is all! You are now ready to create the specification from JSON.

Creating a workflow specification (using JSON)

Now we can use the NuclearStrike in the workflow specification in JSON. Note that this specification is the same as in our first tutorial, except that it references our class strike.NuclearStrike.

{
    "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": "strike.NuclearStrike",
            "name": "nuclear_strike",
            "inputs": [
                "president"
            ]
        },
        "workflow_aborted": {
            "class": "SpiffWorkflow.specs.Cancel.Cancel",
            "name": "workflow_aborted",
            "inputs": [
                "general",
                "president"
            ]
        }
    },
    "description": "",
    "file": null,
    "name": ""
}

Using the custom serializer and task

Here we use our brand new serializer in practice:

from SpiffWorkflow.workflow import Workflow
from SpiffWorkflow.specs.WorkflowSpec import WorkflowSpec
from serializer import NuclearSerializer

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

# 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)