Implementing a Custom Task Spec
Suppose we wanted to manage Timer Start Events outside of SpiffWorkflow. If we have a process loaded up and running that starts with a timer, the timer waits until the event occurs; this might be days or weeks later.
Of course, we can always check that it’s waiting and serialize the workflow until that time. However, we might decide that we don’t want SpiffWorkflow to manage this at all. We could do this with a custom task spec.
The code for this example can be found in misc/custom_start_event.py.
There is a very simple diagram timer_start.bpmn with the process ID timer_start with a Start Event with a Duration Timer of one day that can be used to illustrate how the custom task works. If you run this workflow with any of the configurations provided, you’ll see a WAITING Start Event; if you use the parser and serializer we just created, you’ll be propmted to complete the User Task immediately.
To run this model with the custom spec:
./runner.py -e spiff_example.misc.custom_start_event add -p timer_start -b bpmn/tutorial/timer_start.bpmn
./runner.py -e spiff_example.misc.custom_start_event
First we’ll create a new class.
Note
It might be better have the class’s init method take both the event definition to use and the timer event definition. Unfortunately, our parser is not terribly intuitive or easily extendable, so I’ve done it this way to make this a little easier to follow.
from SpiffWorkflow.bpmn.specs.event_definitions import NoneEventDefinition
from SpiffWorkflow.bpmn.specs.event_definitions.timer import TimerEventDefinition
from SpiffWorkflow.bpmn.specs.mixins import StartEventMixin
from SpiffWorkflow.spiff.specs import SpiffBpmnTask
class CustomStartEvent(StartEventMixin, SpiffBpmnTask):
def __init__(self, wf_spec, bpmn_id, event_definition, **kwargs):
if isinstance(event_definition, TimerEventDefinition):
super().__init__(wf_spec, bpmn_id, NoneEventDefinition(), **kwargs)
self.timer_event = event_definition
else:
super().__init__(wf_spec, bpmn_id, event_definition, **kwargs)
self.timer_event = None
When we create our custom spec, we’ll check to see if we’re creating a Start Event with a TimerEventDefinition
, and
if so, we’ll replace it with a NoneEventDefinition
. There are three different types of Timer Events, so we’ll use
the base class for all three to make sure we account for TimeDate, Duration, and Cycle.
Note
Our class inherits from two classes. We import a mixin class that defines generic BPMN Start Event behavior from
StartEventMixin
in the bpmn
package and the SpiffBpmnTask
from the spiff
package, which
extends the default BpmnSpecMixin
.
We’ve split the basic behavior for specific BPMN tasks from the BpmnSpecMixin
to make it easier to extend
them without running into MRO issues.
In general, if you implement a custom task spec, you’ll need to inherit from bases of both categories.
Whenever we create a custom task spec, we’ll need to create a converter for it so that it can be serialized.
from SpiffWorkflow.bpmn.serializer import BpmnWorkflowSerializer
from SpiffWorkflow.spiff.serializer.task_spec import SpiffBpmnTaskConverter
from SpiffWorkflow.spiff.serializer import DEFAULT_CONFIG
class CustomStartEventConverter(SpiffBpmnTaskConverter):
def to_dict(self, spec):
dct = super().to_dict(spec)
dct['event_definition'] = self.registry.convert(spec.event_definition)
dct['timer_event'] = self.registry.convert(spec.timer_event)
return dct
def from_dict(self, dct):
spec = super().from_dict(dct)
spec.event_definition = self.registry.restore(dct['event_definition'])
spec.timer_event = self.registry.restore(dct['timer_event'])
return spec
Our converter will inherit from the SpiffBpmnTaskConverter
, since that’s our base generic BPMN mixin class.
The parent converter will handle serializing the standard BPMN attributes, as well as attributes added in the
spiff
package. There is a similar base converter in the bpmn.serializer.helpers
package.
A converter needs to implement two methods: to_dict
(which takes a task spec and returns a JSON-serializable
dictionary of its attributes) and from_dict
(which takes the dictionary and returns a task spec of the
appropriate type. We call the base method to do most of the work, and then update the result to reflect the changes
we made, in this case ensuring that both event definitions are handled. The parent converter also provides convert
and restore
methods to serialize any object that Spiff’s serializer knows how to handle. For more details about
the serializer, see Introduction.
When we create our serializer, we need to tell it about this task. The serializer is initialized with a mapping of object class to converter class, so we just need to add an entry for this mapping.
SPIFF_CONFIG[CustomStartEvent] = CustomStartEventConverter
registry = FileSerializer.configure(SPIFF_CONFIG)
serializer = FileSerializer(dirname, registry=registry)
We also have to tell the parser to use our class instead of the standard class.
from SpiffWorkflow.spiff.parser import SpiffBpmnParser
from SpiffWorkflow.spiff.parser.event_parsers import StartEventParser
from SpiffWorkflow.bpmn.parser.util import full_tag
parser = SpiffBpmnParser()
parser.OVERRIDE_PARSER_CLASSES[full_tag('startEvent')] = (StartEventParser, CustomStartEvent)
The parser contains class attributes that define how to parse a particular element and the class that should be used to create the task spec, so rather than pass these in as arguments, we create a parser and then update the values it will use. This is a bit unintuitive, but that’s how it works.
Fortunately, we were able to reuse an existing Task Spec parser, which simplifies the process quite a bit.