1
|
|
|
# -*- coding: utf-8 -*- |
2
|
|
|
# |
3
|
|
|
# This file is part of SENAITE.CORE. |
4
|
|
|
# |
5
|
|
|
# SENAITE.CORE is free software: you can redistribute it and/or modify it under |
6
|
|
|
# the terms of the GNU General Public License as published by the Free Software |
7
|
|
|
# Foundation, version 2. |
8
|
|
|
# |
9
|
|
|
# This program is distributed in the hope that it will be useful, but WITHOUT |
10
|
|
|
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS |
11
|
|
|
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more |
12
|
|
|
# details. |
13
|
|
|
# |
14
|
|
|
# You should have received a copy of the GNU General Public License along with |
15
|
|
|
# this program; if not, write to the Free Software Foundation, Inc., 51 |
16
|
|
|
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. |
17
|
|
|
# |
18
|
|
|
# Copyright 2018-2024 by it's authors. |
19
|
|
|
# Some rights reserved, see README and LICENSE. |
20
|
|
|
|
21
|
|
|
from bika.lims import api |
22
|
|
|
from bika.lims.api import _marker |
23
|
|
|
from Products.DCWorkflow.DCWorkflow import DCWorkflowDefinition |
24
|
|
|
from Products.DCWorkflow.Guard import Guard |
25
|
|
|
from Products.DCWorkflow.States import StateDefinition |
26
|
|
|
from Products.DCWorkflow.Transitions import TransitionDefinition |
27
|
|
|
from senaite.core import logger |
28
|
|
|
|
29
|
|
|
|
30
|
|
|
def get_workflow(thing, default=_marker): |
31
|
|
|
"""Returns the primary DCWorkflowDefinition object for the thing passed-in |
32
|
|
|
|
33
|
|
|
:param thing: A single catalog brain, content object, supermodel, workflow, |
34
|
|
|
workflow id, workflow state, workflow transition or portal type |
35
|
|
|
:type thing: DCWorkflowDefinition/StateDefinition/TransitionDefinition/ |
36
|
|
|
ATContentType/DexterityContentType/CatalogBrain/SuperModel/string |
37
|
|
|
:return: The primary workflow of the thing |
38
|
|
|
:rtype: Products.DCWorkflow.DCWorkflow.DCWorkflowDefinition |
39
|
|
|
""" |
40
|
|
|
if isinstance(thing, DCWorkflowDefinition): |
41
|
|
|
return thing |
42
|
|
|
if isinstance(thing, StateDefinition): |
43
|
|
|
return thing.getWorkflow() |
44
|
|
|
if isinstance(thing, TransitionDefinition): |
45
|
|
|
return thing.getWorkflow() |
46
|
|
|
|
47
|
|
|
if api.is_string(thing): |
48
|
|
|
# Look-up the workflow by id |
49
|
|
|
wf_tool = api.get_tool("portal_workflow") |
50
|
|
|
workflow = wf_tool.getWorkflowById(thing) |
51
|
|
|
if workflow: |
52
|
|
|
return workflow |
53
|
|
|
|
54
|
|
|
if api.is_string(thing) or api.is_object(thing): |
55
|
|
|
# Look-up the workflow by portal type or object |
56
|
|
|
wf_tool = api.get_tool("portal_workflow") |
57
|
|
|
workflows = wf_tool.getChainFor(thing) |
58
|
|
|
if len(workflows) == 1: |
59
|
|
|
return wf_tool.getWorkflowById(workflows[0]) |
60
|
|
|
if default is not _marker: |
61
|
|
|
if default is None: |
62
|
|
|
return default |
63
|
|
|
return get_workflow(default) |
64
|
|
|
if len(workflows) > 1: |
65
|
|
|
raise ValueError("More than one workflow: %s" % repr(thing)) |
66
|
|
|
raise ValueError("Workflow not found: %s" % repr(thing)) |
67
|
|
|
|
68
|
|
|
if default is not _marker: |
69
|
|
|
if default is None: |
70
|
|
|
return default |
71
|
|
|
return get_workflow(default) |
72
|
|
|
|
73
|
|
|
raise ValueError("Type is not supported: %s" % repr(type(thing))) |
74
|
|
|
|
75
|
|
|
|
76
|
|
|
def get_state(workflow, state_id, default=_marker): |
77
|
|
|
"""Returns the workflow state with the given id |
78
|
|
|
:param workflow: Workflow object or workflow id |
79
|
|
|
:type workflow: DCWorkflowDefinition/string |
80
|
|
|
:param state_id: Workflow state id |
81
|
|
|
:type state_id: string |
82
|
|
|
:return: The state object for the given workflow and id |
83
|
|
|
:rtype: Products.DCWorkflow.States.StateDefinition |
84
|
|
|
""" |
85
|
|
|
wf = get_workflow(workflow) |
86
|
|
|
state = wf.states.get(state_id) |
87
|
|
|
if state: |
88
|
|
|
return state |
89
|
|
|
if default is not _marker: |
90
|
|
|
return default |
91
|
|
|
raise ValueError("State %s not found for %s" % (state_id, wf.id)) |
92
|
|
|
|
93
|
|
|
|
94
|
|
|
def get_transition(workflow, transition_id, default=_marker): |
95
|
|
|
"""Returns the workflow transition with the given id |
96
|
|
|
:param workflow: Workflow object or workflow id |
97
|
|
|
:type workflow: DCWorkflowDefinition/string |
98
|
|
|
:param transition_id: Workflow transition id |
99
|
|
|
:type transition_id: string |
100
|
|
|
:return: The transition object for the given workflow and id |
101
|
|
|
:rtype: Products.DCWorkflow.Transitions.TransitionDefinition |
102
|
|
|
""" |
103
|
|
|
wf = get_workflow(workflow) |
104
|
|
|
transition = wf.transitions.get(transition_id) |
105
|
|
|
if transition: |
106
|
|
|
return transition |
107
|
|
|
if default is not _marker: |
108
|
|
|
return default |
109
|
|
|
raise ValueError("Transition %s not found for %s" % (transition_id, wf.id)) |
110
|
|
|
|
111
|
|
|
|
112
|
|
|
def update_workflow(workflow, states=None, transitions=None, **kwargs): |
113
|
|
|
"""Updates an existing workflow |
114
|
|
|
|
115
|
|
|
Usage:: |
116
|
|
|
|
117
|
|
|
>>> from senaite.core import permissions |
118
|
|
|
>>> from senaite.core.workflow import SAMPLE_WORKFLOW |
119
|
|
|
>>> states = { |
120
|
|
|
... "stored": { |
121
|
|
|
... "title": "Stored", |
122
|
|
|
... "description": "Sample is stored", |
123
|
|
|
... # Use tuples to overwrite existing transitions. To extend |
124
|
|
|
... # existing transitions, use a list |
125
|
|
|
... "transitions": ("recover", "detach", "dispatch", ), |
126
|
|
|
... # Copy permissions from sample_received first |
127
|
|
|
... "permissions_copy_from": "sample_received", |
128
|
|
|
... # Permissions mapping |
129
|
|
|
... "permissions": { |
130
|
|
|
... # Use tuples to overwrite existing and acquire=False. |
131
|
|
|
... # To extend existing roles, use a list |
132
|
|
|
... permissions.TransitionCancelAnalysisRequest: (), |
133
|
|
|
... permissions.TransitionReinstateAnalysisRequest: (), |
134
|
|
|
... } |
135
|
|
|
... }, |
136
|
|
|
... } |
137
|
|
|
>>> trans = { |
138
|
|
|
... "store": { |
139
|
|
|
... "title": "Store", |
140
|
|
|
... "new_state": "stored", |
141
|
|
|
... "action": "Store sample", |
142
|
|
|
... "guard": { |
143
|
|
|
... "guard_permissions": "", |
144
|
|
|
... "guard_roles": "", |
145
|
|
|
... "guard_expr": "python:here.guard_handler('store')", |
146
|
|
|
... } |
147
|
|
|
... }, |
148
|
|
|
... } |
149
|
|
|
>>> update_workflow(SAMPLE_WORKFLOW, states=states, transitions=trans) |
150
|
|
|
|
151
|
|
|
:param workflow: Workflow object or workflow id |
152
|
|
|
:type workflow: DCWorkflowDefinition/string |
153
|
|
|
:param states: states to be updated/created |
154
|
|
|
:type states: dict of {state_id:{<state_properties>}} |
155
|
|
|
:param transitions: transitions to be updated/created |
156
|
|
|
:type transitions: dict of {transition_id:<transition_properties>} |
157
|
|
|
:param title: (optional) the title of the workflow or None |
158
|
|
|
:type title: string |
159
|
|
|
:param description: (optional) the description of the workflow or None |
160
|
|
|
:type description: string |
161
|
|
|
:param initial_state: (optional) the initial status id of the workflow |
162
|
|
|
:type initial_state: string |
163
|
|
|
""" |
164
|
|
|
wf = get_workflow(workflow) |
165
|
|
|
|
166
|
|
|
# Set basic info (title, description, etc.) |
167
|
|
|
wf.title = kwargs.get("title", wf.title) |
168
|
|
|
wf.description = kwargs.get("description", wf.description) |
169
|
|
|
wf.initial_state = kwargs.get("initial_state", wf.initial_state) |
170
|
|
|
|
171
|
|
|
# Update states |
172
|
|
|
states = states or {} |
173
|
|
|
for state_id, values in states.items(): |
174
|
|
|
|
175
|
|
|
# Create the state if it does not exist yet |
176
|
|
|
state = wf.states.get(state_id) |
177
|
|
|
if not state: |
178
|
|
|
wf.states.addState(state_id) |
179
|
|
|
state = wf.states.get(state_id) |
180
|
|
|
|
181
|
|
|
# Update the state with the settings passed-in |
182
|
|
|
update_state(state, **values) |
183
|
|
|
|
184
|
|
|
# Update transitions |
185
|
|
|
transitions = transitions or {} |
186
|
|
|
for transition_id, values in transitions.items(): |
187
|
|
|
|
188
|
|
|
transition = wf.transitions.get(transition_id) |
189
|
|
|
if not transition: |
190
|
|
|
wf.transitions.addTransition(transition_id) |
191
|
|
|
transition = wf.transitions.get(transition_id) |
192
|
|
|
|
193
|
|
|
# Update the transition with the settings passed-in |
194
|
|
|
update_transition(transition, **values) |
195
|
|
|
|
196
|
|
|
|
197
|
|
|
def update_state(state, transitions=None, permissions=None, **kwargs): |
198
|
|
|
"""Updates the state of an existing workflow |
199
|
|
|
|
200
|
|
|
Note that regarding the use of tuples/lists for roles in permissions and |
201
|
|
|
in transitions, the same principles from DCWorkflow apply. This is: |
202
|
|
|
|
203
|
|
|
- Transitions passed-in as a list extend the existing ones |
204
|
|
|
- Transitions passed-in as a tuple replace the existing ones |
205
|
|
|
- Roles passed-in as a list extend the existing ones and acquire is kept |
206
|
|
|
- Roles passed-in as a tuple replace the existing ones and acquire is '0' |
207
|
|
|
|
208
|
|
|
Usage:: |
209
|
|
|
|
210
|
|
|
>>> from senaite.core import permissions |
211
|
|
|
>>> from senaite.core.workflow import SAMPLE_WORKFLOW |
212
|
|
|
|
213
|
|
|
>>> # Use tuples to overwrite existing transitions. To extend |
214
|
|
|
>>> # existing transitions, use a list |
215
|
|
|
>>> trans = ("recover", "detach", "dispatch", ) |
216
|
|
|
|
217
|
|
|
>>> # Use tuples to override existing roles per permission and to also |
218
|
|
|
>>> # set acquire to False. To extend existing roles and preserve |
219
|
|
|
>>> # acquire, use a list |
220
|
|
|
>>> perms = { |
221
|
|
|
... permissions.TransitionCancelAnalysisRequest: (), |
222
|
|
|
... permissions.TransitionReinstateAnalysisRequest: (), |
223
|
|
|
... } |
224
|
|
|
|
225
|
|
|
>>> kwargs = { |
226
|
|
|
... "title": "Stored", |
227
|
|
|
... "description": "Sample is stored", |
228
|
|
|
... # Copy permissions from sample_received first |
229
|
|
|
... "permissions_copy_from": "sample_received", |
230
|
|
|
... } |
231
|
|
|
|
232
|
|
|
>>> state = get_state(SAMPLE_WORKFLOW, "stored") |
233
|
|
|
>>> update_state(state, transitions=trans, permissions=perms, **kwargs) |
234
|
|
|
|
235
|
|
|
:param state: Workflow state definition object |
236
|
|
|
:type state: Products.DCWorkflow.States.StateDefinition |
237
|
|
|
:param transitions: Tuple or list of ids from transitions to be considered |
238
|
|
|
as exit transitions of the state. If a tuple, existing transitions are |
239
|
|
|
replaced by the new ones. If a list, existing transitions are extended |
240
|
|
|
with the new ones. If None, keeps the original transitions. |
241
|
|
|
:type transitions: list[string] |
242
|
|
|
:param permissions: dict of {permission_id:roles} where 'roles' can be a |
243
|
|
|
tuple or a list. If a tuple, existing roles are replaced by new ones |
244
|
|
|
and acquired is set to 'False'. If a list, existing roles are extended |
245
|
|
|
with the new ones and acquired is not changed. If None, keeps the |
246
|
|
|
original permissions |
247
|
|
|
:type: permissions: dict({string:tuple|list}) |
248
|
|
|
:param title: (optional) the title of the workflow or None |
249
|
|
|
:type title: string |
250
|
|
|
:param description: (optional) the description of the workflow or None |
251
|
|
|
:type description: string |
252
|
|
|
:param permissions_copy_from: (optional) the id of the status to copy the |
253
|
|
|
permissions from to the given status. When set, the permissions from |
254
|
|
|
the source status are copied to the given status before the update of |
255
|
|
|
the permissions with those passed-in takes place. |
256
|
|
|
:type: permissions_copy_from: string |
257
|
|
|
""" |
258
|
|
|
# set basic info (title, description, etc.) |
259
|
|
|
state.title = kwargs.get("title", state.title) |
260
|
|
|
state.description = kwargs.get("description", state.description) |
261
|
|
|
|
262
|
|
|
# check if we need to replace or extend existing transitions |
263
|
|
|
if transitions is None: |
264
|
|
|
transitions = state.transitions |
265
|
|
|
elif isinstance(transitions, list): |
266
|
|
|
transitions = set(transitions) |
267
|
|
|
transitions.update(state.transitions) |
268
|
|
|
transitions = tuple(transitions) |
269
|
|
|
|
270
|
|
|
# set transitions |
271
|
|
|
state.transitions = transitions |
272
|
|
|
|
273
|
|
|
# copy permissions fromm another state |
274
|
|
|
source = kwargs.get("permissions_copy_from") |
275
|
|
|
if source: |
276
|
|
|
wf = get_workflow(state) |
277
|
|
|
source = wf.states.get(source) |
278
|
|
|
copy_permissions(source, state) |
279
|
|
|
|
280
|
|
|
# update existing permissions |
281
|
|
|
permissions = permissions or {} |
282
|
|
|
for perm_id, roles in permissions.items(): |
283
|
|
|
update_permission(state, perm_id, roles) |
284
|
|
|
|
285
|
|
|
|
286
|
|
|
def update_transition(transition, **kwargs): |
287
|
|
|
"""Updates a workflow transition |
288
|
|
|
|
289
|
|
|
Usage:: |
290
|
|
|
|
291
|
|
|
>>> from senaite.core.workflow import SAMPLE_WORKFLOW |
292
|
|
|
|
293
|
|
|
>>> guard = { |
294
|
|
|
... "guard_permissions": "", |
295
|
|
|
... "guard_roles": "", |
296
|
|
|
... "guard_expr": "python:here.guard_handler('store')", |
297
|
|
|
... } |
298
|
|
|
|
299
|
|
|
>>> wf = get_workflow(SAMPLE_WORKFLOW) |
300
|
|
|
>>> transition = wf.transitions.get("store") |
301
|
|
|
>>> update_transition(transition, |
302
|
|
|
... title="Store", |
303
|
|
|
... description="The action to store the sample", |
304
|
|
|
... action="Store sample", |
305
|
|
|
... new_state="stored", |
306
|
|
|
... guard=guard) |
307
|
|
|
|
308
|
|
|
:param transition: Workflow transition definition object |
309
|
|
|
:type transition: Products.DCWorkflow.Transitions.TransitionDefinition |
310
|
|
|
:param title: (optional) the title of the transition |
311
|
|
|
:type title: string |
312
|
|
|
:param description: (optional) the descrioption of the transition |
313
|
|
|
:type title: string |
314
|
|
|
:param new_state: (optional) the state of the object after the transition |
315
|
|
|
:type new_state: string |
316
|
|
|
:param after_script: (optional) Script (Python) to run after the transition |
317
|
|
|
:type after_script: string |
318
|
|
|
:param action: (optional) the action name to display in the actions box |
319
|
|
|
:type action: string |
320
|
|
|
:param action_url: (optional) the url to use for this action. The |
321
|
|
|
%(content_url) wildcard is replaced by absolute path at runtime. If |
322
|
|
|
empty, the default `content_status_modify?workflow_action=<action_id>` |
323
|
|
|
will be used at runtime |
324
|
|
|
:type action_url: string |
325
|
|
|
:param guard: (optional) a dict with Guard properties. The supported |
326
|
|
|
properties are: 'guard_roles', 'guard_groups', 'guard_expr' and |
327
|
|
|
'guard_permissions' |
328
|
|
|
:type guard: dict |
329
|
|
|
""" |
330
|
|
|
# attrs conversions |
331
|
|
|
mapping = { |
332
|
|
|
"title": "title", |
333
|
|
|
"description": "description", |
334
|
|
|
"action": "actbox_name", |
335
|
|
|
"action_url": "actbox_url", |
336
|
|
|
"after_script": "after_script_name", |
337
|
|
|
"new_state": "new_state_id", |
338
|
|
|
} |
339
|
|
|
properties = {} |
340
|
|
|
for key, property_id in mapping.items(): |
341
|
|
|
default = getattr(transition, property_id) |
342
|
|
|
value = kwargs.get(key, None) |
343
|
|
|
if value is None: |
344
|
|
|
value = default |
345
|
|
|
properties[property_id] = value |
346
|
|
|
|
347
|
|
|
# update the transition properties |
348
|
|
|
transition.setProperties(**properties) |
349
|
|
|
|
350
|
|
|
# update the guard |
351
|
|
|
guard = transition.guard or Guard() |
352
|
|
|
guard_props = kwargs.get("guard") |
353
|
|
|
if guard_props: |
354
|
|
|
guard.changeFromProperties(guard_props) |
355
|
|
|
transition.guard = guard |
356
|
|
|
|
357
|
|
|
|
358
|
|
|
def update_permission(state, permission_id, roles): |
359
|
|
|
"""Updates the permission mappings of an existing workflow state |
360
|
|
|
|
361
|
|
|
:param state: Workflow state definition object |
362
|
|
|
:type state: Products.DCWorkflow.States.StateDefinition |
363
|
|
|
:param permission_id: id of the permission |
364
|
|
|
:type permission_id: string |
365
|
|
|
:param roles: List or tuple with roles to which the given permission has |
366
|
|
|
to be granted for the workflow status. If a tuple, acquire is set to |
367
|
|
|
False and roles overwritten. Thus, an empty tuple clears all roles for |
368
|
|
|
this permission and state. If a list, the existing roles are extended |
369
|
|
|
with new ones and acquire setting is not modified. |
370
|
|
|
:type roles: list/tuple |
371
|
|
|
""" |
372
|
|
|
# resolve acquire |
373
|
|
|
if isinstance(roles, tuple): |
374
|
|
|
acquired = 0 |
375
|
|
|
else: |
376
|
|
|
info = state.getPermissionInfo(permission_id) or {} |
377
|
|
|
acquired = info.get("acquired", 1) |
378
|
|
|
roles = set(roles) |
379
|
|
|
roles.update(info.get("roles", [])) |
380
|
|
|
|
381
|
|
|
# sort them for human-friendly reading on retrieval |
382
|
|
|
roles = tuple(sorted(roles)) |
383
|
|
|
|
384
|
|
|
# add this permission to the workflow if not globally defined yet |
385
|
|
|
wf = get_workflow(state) |
386
|
|
|
if permission_id not in wf.permissions: |
387
|
|
|
wf.permissions = wf.permissions + (permission_id,) |
388
|
|
|
|
389
|
|
|
# set the permission |
390
|
|
|
logger.info("{}.{}: '{}' (acquired={}): '{}'".format( |
391
|
|
|
wf.id, state.id, permission_id, repr(acquired), ', '.join(roles))) |
392
|
|
|
state.setPermission(permission_id, acquired, roles) |
393
|
|
|
|
394
|
|
|
|
395
|
|
|
def copy_permissions(source, destination): |
396
|
|
|
"""Copies the permission mappings of a workflow state to another |
397
|
|
|
|
398
|
|
|
:param source: Workflow state definition object used as source |
399
|
|
|
:type source: Products.DCWorkflow.States.StateDefinition |
400
|
|
|
:param destination: Workflow state definition object used as destination |
401
|
|
|
:type destination: Products.DCWorkflow.States.StateDefinition |
402
|
|
|
""" |
403
|
|
|
for permission in source.permissions: |
404
|
|
|
info = source.getPermissionInfo(permission) |
405
|
|
|
roles = info.get("roles") or [] |
406
|
|
|
acquired = info.get("acquired", 1) |
407
|
|
|
# update the roles for this permission at destination |
408
|
|
|
destination.setPermission(permission, acquired, sorted(roles)) |
409
|
|
|
|