-
Notifications
You must be signed in to change notification settings - Fork 310
/
task_spec.py
120 lines (103 loc) · 4.95 KB
/
task_spec.py
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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
from lxml import etree
from SpiffWorkflow.dmn.specs.BusinessRuleTask import BusinessRuleTask
from SpiffWorkflow.bpmn.parser.TaskParser import TaskParser
from SpiffWorkflow.bpmn.parser.task_parsers import SubprocessParser
from SpiffWorkflow.bpmn.parser.util import xpath_eval
SPIFFWORKFLOW_MODEL_NS = 'http://spiffworkflow.org/bpmn/schema/1.0/core'
class SpiffTaskParser(TaskParser):
def parse_extensions(self, node=None):
# Too bad doing this works in such a stupid way.
# We should set a namespace and automatically do this.
extensions = {}
extra_ns = {'spiffworkflow': SPIFFWORKFLOW_MODEL_NS}
xpath = xpath_eval(self.node, extra_ns) if node is None else xpath_eval(node, extra_ns)
extension_nodes = xpath('.//bpmn:extensionElements/spiffworkflow:*')
for node in extension_nodes:
name = etree.QName(node).localname
if name == 'properties':
extensions['properties'] = self._parse_properties(node)
elif name == 'serviceTaskOperator':
extensions['serviceTaskOperator'] = self._parse_servicetask_operator(node)
else:
extensions[name] = node.text
return extensions
def _parse_properties(self, node):
extra_ns = {'spiffworkflow': SPIFFWORKFLOW_MODEL_NS}
xpath = xpath_eval(node, extra_ns)
property_nodes = xpath('.//spiffworkflow:property')
properties = {}
for prop_node in property_nodes:
properties[prop_node.attrib['name']] = prop_node.attrib['value']
return properties
def _parse_servicetask_operator(self, node):
name = node.attrib['id']
extra_ns = {'spiffworkflow': SPIFFWORKFLOW_MODEL_NS}
xpath = xpath_eval(node, extra_ns)
parameter_nodes = xpath('.//spiffworkflow:parameter')
operator = {'name': name}
parameters = {}
for param_node in parameter_nodes:
if 'value' in param_node.attrib:
parameters[param_node.attrib['id']] = {
'value': param_node.attrib['value'],
'type': param_node.attrib['type']
}
operator['parameters'] = parameters
return operator
def create_task(self):
# The main task parser already calls this, and even sets an attribute, but
# 1. It calls it after creating the task so I don't have access to it here yet and
# 2. I want defined attributes, not a dict of random crap
# (though the dict of random crap will still be there since the base parser adds it).
extensions = self.parse_extensions()
prescript = extensions.get('preScript')
postscript = extensions.get('postScript')
return self.spec_class(self.spec, self.get_task_spec_name(),
lane=self.lane,
description=self.node.get('name', None),
position=self.position,
prescript=prescript,
postscript=postscript)
class SubWorkflowParser(SpiffTaskParser):
def create_task(self):
extensions = self.parse_extensions()
prescript = extensions.get('preScript')
postscript = extensions.get('postScript')
subworkflow_spec = SubprocessParser.get_subprocess_spec(self)
return self.spec_class(
self.spec, self.get_task_spec_name(), subworkflow_spec,
lane=self.lane, position=self.position,
description=self.node.get('name', None),
prescript=prescript,
postscript=postscript)
class CallActivityParser(SpiffTaskParser):
def create_task(self):
extensions = self.parse_extensions()
prescript = extensions.get('preScript')
postscript = extensions.get('postScript')
subworkflow_spec = SubprocessParser.get_call_activity_spec(self)
return self.spec_class(
self.spec, self.get_task_spec_name(), subworkflow_spec,
lane=self.lane, position=self.position,
description=self.node.get('name', None),
prescript=prescript,
postscript=postscript)
class ServiceTaskParser(SpiffTaskParser):
def create_task(self):
extensions = self.parse_extensions()
operator = extensions.get('serviceTaskOperator')
return self.spec_class(
self.spec, self.get_task_spec_name(),
operator['name'], operator['parameters'],
lane=self.lane, position=self.position)
class BusinessRuleTaskParser(SpiffTaskParser):
def create_task(self):
extensions = self.parse_extensions()
decision_ref = extensions.get('calledDecisionId')
return BusinessRuleTask(self.spec,
self.get_task_spec_name(),
dmnEngine=self.process_parser.parser.get_engine(decision_ref, self.node),
lane=self.lane,
position=self.position,
description=self.node.get('name', None)
)