spiff-example-cli/spiff_example/misc/custom_start_event.py at dc5be6875487d05601d225ef78f988e5fa4b0aee · sartography/spiff-example-cli · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
import logging

from RestrictedPython import safe_globals

from SpiffWorkflow.spiff.parser import SpiffBpmnParser
from SpiffWorkflow.spiff.specs.defaults import UserTask, ManualTask
from SpiffWorkflow.spiff.serializer.config import SPIFF_CONFIG
from SpiffWorkflow.bpmn.specs.mixins.none_task import NoneTask
from SpiffWorkflow.bpmn.script_engine import TaskDataEnvironment

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

from SpiffWorkflow.spiff.parser.event_parsers import StartEventParser
from SpiffWorkflow.bpmn.parser.util import full_tag
from SpiffWorkflow.bpmn.serializer.default import EventConverter
from SpiffWorkflow.spiff.serializer.task_spec import SpiffBpmnTaskConverter

from ..serializer.file import FileSerializer
from ..engine import BpmnEngine
from .curses_handlers import UserTaskHandler, ManualTaskHandler

logger = logging.getLogger('spiff_engine')
logger.setLevel(logging.INFO)

spiff_logger = logging.getLogger('spiff')
spiff_logger.setLevel(logging.INFO)

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

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

dirname = 'data'
FileSerializer.initialize(dirname)

SPIFF_CONFIG[CustomStartEvent] = CustomStartEventConverter

registry = FileSerializer.configure(SPIFF_CONFIG)
serializer = FileSerializer(dirname, registry=registry)

parser = SpiffBpmnParser()
parser.OVERRIDE_PARSER_CLASSES[full_tag('startEvent')] = (StartEventParser, CustomStartEvent)

handlers = {
    UserTask: UserTaskHandler,
    ManualTask: ManualTaskHandler,
    NoneTask: ManualTaskHandler,
}

script_env = TaskDataEnvironment(safe_globals)

engine = BpmnEngine(parser, serializer, script_env)