senaite.core.api.workflow.get_transition() - Code Metrics - Inspection of "Functions for workflow update (#2514)" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — 2.x ( 1c5375...9d7a5b )

by Ramon

created 2024-03-18 12:28 UTC

senaite.core.api.workflow.get_transition() A

↳ Parent: senaite.core.api.workflow

Complexity

Conditions

Size

Total Lines	16
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	8
dl	0
loc	16
rs	10
c	0
b	0
f	0
cc	3
nop	3

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE.
#
# SENAITE.CORE is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, version 2.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright 2018-2024 by it's authors.
# Some rights reserved, see README and LICENSE.

from bika.lims import api
from bika.lims.api import _marker
from Products.DCWorkflow.DCWorkflow import DCWorkflowDefinition
from Products.DCWorkflow.Guard import Guard
from Products.DCWorkflow.States import StateDefinition
from Products.DCWorkflow.Transitions import TransitionDefinition
from senaite.core import logger


def get_workflow(thing, default=_marker):
    """Returns the primary DCWorkflowDefinition object for the thing passed-in

    :param thing: A single catalog brain, content object, supermodel, workflow,
        workflow id, workflow state, workflow transition or portal type
    :type thing: DCWorkflowDefinition/StateDefinition/TransitionDefinition/
        ATContentType/DexterityContentType/CatalogBrain/SuperModel/string
    :return: The primary workflow of the thing
    :rtype: Products.DCWorkflow.DCWorkflow.DCWorkflowDefinition
    """
    if isinstance(thing, DCWorkflowDefinition):
        return thing
    if isinstance(thing, StateDefinition):
        return thing.getWorkflow()
    if isinstance(thing, TransitionDefinition):
        return thing.getWorkflow()

    if api.is_string(thing):
        # Look-up the workflow by id
        wf_tool = api.get_tool("portal_workflow")
        workflow = wf_tool.getWorkflowById(thing)
        if workflow:
            return workflow

    if api.is_string(thing) or api.is_object(thing):
        # Look-up the workflow by portal type or object
        wf_tool = api.get_tool("portal_workflow")
        workflows = wf_tool.getChainFor(thing)
        if len(workflows) == 1:
            return wf_tool.getWorkflowById(workflows[0])
        if default is not _marker:
            if default is None:
                return default
            return get_workflow(default)
        if len(workflows) > 1:
            raise ValueError("More than one workflow: %s" % repr(thing))
        raise ValueError("Workflow not found: %s" % repr(thing))

    if default is not _marker:
        if default is None:
            return default
        return get_workflow(default)

    raise ValueError("Type is not supported: %s" % repr(type(thing)))


def get_state(workflow, state_id, default=_marker):
    """Returns the workflow state with the given id
    :param workflow: Workflow object or workflow id
    :type workflow: DCWorkflowDefinition/string
    :param state_id: Workflow state id
    :type state_id: string
    :return: The state object for the given workflow and id
    :rtype: Products.DCWorkflow.States.StateDefinition
    """
    wf = get_workflow(workflow)
    state = wf.states.get(state_id)
    if state:
        return state
    if default is not _marker:
        return default
    raise ValueError("State %s not found for %s" % (state_id, wf.id))


def get_transition(workflow, transition_id, default=_marker):
    """Returns the workflow transition with the given id
    :param workflow: Workflow object or workflow id
    :type workflow: DCWorkflowDefinition/string
    :param transition_id: Workflow transition id
    :type transition_id: string
    :return: The transition object for the given workflow and id
    :rtype: Products.DCWorkflow.Transitions.TransitionDefinition
    """
    wf = get_workflow(workflow)
    transition = wf.transitions.get(transition_id)
    if transition:
        return transition
    if default is not _marker:
        return default
    raise ValueError("Transition %s not found for %s" % (transition_id, wf.id))


def update_workflow(workflow, states=None, transitions=None, **kwargs):
    """Updates an existing workflow

    Usage::

        >>> from senaite.core import permissions
        >>> from senaite.core.workflow import SAMPLE_WORKFLOW
        >>> states = {
        ...     "stored": {
        ...         "title": "Stored",
        ...         "description": "Sample is stored",
        ...         # Use tuples to overwrite existing transitions. To extend
        ...         # existing transitions, use a list
        ...         "transitions": ("recover", "detach", "dispatch", ),
        ...         # Copy permissions from sample_received first
        ...         "permissions_copy_from": "sample_received",
        ...         # Permissions mapping
        ...         "permissions": {
        ...             # Use tuples to overwrite existing and acquire=False.
        ...             # To extend existing roles, use a list
        ...             permissions.TransitionCancelAnalysisRequest: (),
        ...             permissions.TransitionReinstateAnalysisRequest: (),
        ...         }
        ...     },
        ... }
        >>> trans = {
        ...     "store": {
        ...         "title": "Store",
        ...         "new_state": "stored",
        ...         "action": "Store sample",
        ...         "guard": {
        ...             "guard_permissions": "",
        ...             "guard_roles": "",
        ...             "guard_expr": "python:here.guard_handler('store')",
        ...         }
        ...     },
        ... }
        >>> update_workflow(SAMPLE_WORKFLOW, states=states, transitions=trans)

    :param workflow: Workflow object or workflow id
    :type workflow: DCWorkflowDefinition/string
    :param states: states to be updated/created
    :type states: dict of {state_id:{<state_properties>}}
    :param transitions: transitions to be updated/created
    :type transitions: dict of {transition_id:<transition_properties>}
    :param title: (optional) the title of the workflow or None
    :type title: string
    :param description: (optional) the description of the workflow or None
    :type description: string
    :param initial_state: (optional) the initial status id of the workflow
    :type initial_state: string
    """
    wf = get_workflow(workflow)

    # Set basic info (title, description, etc.)
    wf.title = kwargs.get("title", wf.title)
    wf.description = kwargs.get("description", wf.description)
    wf.initial_state = kwargs.get("initial_state", wf.initial_state)

    # Update states
    states = states or {}
    for state_id, values in states.items():

        # Create the state if it does not exist yet
        state = wf.states.get(state_id)
        if not state:
            wf.states.addState(state_id)
            state = wf.states.get(state_id)

        # Update the state with the settings passed-in
        update_state(state, **values)

    # Update transitions
    transitions = transitions or {}
    for transition_id, values in transitions.items():

        transition = wf.transitions.get(transition_id)
        if not transition:
            wf.transitions.addTransition(transition_id)
            transition = wf.transitions.get(transition_id)

        # Update the transition with the settings passed-in
        update_transition(transition, **values)


def update_state(state, transitions=None, permissions=None, **kwargs):
    """Updates the state of an existing workflow

    Note that regarding the use of tuples/lists for roles in permissions and
    in transitions, the same principles from DCWorkflow apply. This is:

    - Transitions passed-in as a list extend the existing ones
    - Transitions passed-in as a tuple replace the existing ones
    - Roles passed-in as a list extend the existing ones and acquire is kept
    - Roles passed-in as a tuple replace the existing ones and acquire is '0'

    Usage::

        >>> from senaite.core import permissions
        >>> from senaite.core.workflow import SAMPLE_WORKFLOW

        >>> # Use tuples to overwrite existing transitions. To extend
        >>> # existing transitions, use a list
        >>> trans = ("recover", "detach", "dispatch", )

        >>> # Use tuples to override existing roles per permission and to also
        >>> # set acquire to False. To extend existing roles and preserve
        >>> # acquire, use a list
        >>> perms = {
        ...     permissions.TransitionCancelAnalysisRequest: (),
        ...     permissions.TransitionReinstateAnalysisRequest: (),
        ... }

        >>> kwargs = {
        ...     "title": "Stored",
        ...     "description": "Sample is stored",
        ...     # Copy permissions from sample_received first
        ...     "permissions_copy_from": "sample_received",
        ... }

        >>> state = get_state(SAMPLE_WORKFLOW, "stored")
        >>> update_state(state, transitions=trans, permissions=perms, **kwargs)

    :param state: Workflow state definition object
    :type state: Products.DCWorkflow.States.StateDefinition
    :param transitions: Tuple or list of ids from transitions to be considered
        as exit transitions of the state. If a tuple, existing transitions are
        replaced by the new ones. If a list, existing transitions are extended
        with the new ones. If None, keeps the original transitions.
    :type transitions: list[string]
    :param permissions: dict of {permission_id:roles} where 'roles' can be a
        tuple or a list. If a tuple, existing roles are replaced by new ones
        and acquired is set to 'False'. If a list, existing roles are extended
        with the new ones and acquired is not changed. If None, keeps the
        original permissions
    :type: permissions: dict({string:tuple|list})
    :param title: (optional) the title of the workflow or None
    :type title: string
    :param description: (optional) the description of the workflow or None
    :type description: string
    :param permissions_copy_from: (optional) the id of the status to copy the
        permissions from to the given status. When set, the permissions from
        the source status are copied to the given status before the update of
        the permissions with those passed-in takes place.
    :type: permissions_copy_from: string
    """
    # set basic info (title, description, etc.)
    state.title = kwargs.get("title", state.title)
    state.description = kwargs.get("description", state.description)

    # check if we need to replace or extend existing transitions
    if transitions is None:
        transitions = state.transitions
    elif isinstance(transitions, list):
        transitions = set(transitions)
        transitions.update(state.transitions)
        transitions = tuple(transitions)

    # set transitions
    state.transitions = transitions

    # copy permissions fromm another state
    source = kwargs.get("permissions_copy_from")
    if source:
        wf = get_workflow(state)
        source = wf.states.get(source)
        copy_permissions(source, state)

    # update existing permissions
    permissions = permissions or {}
    for perm_id, roles in permissions.items():
        update_permission(state, perm_id, roles)


def update_transition(transition, **kwargs):
    """Updates a workflow transition

    Usage::

        >>> from senaite.core.workflow import SAMPLE_WORKFLOW

        >>> guard = {
        ...     "guard_permissions": "",
        ...     "guard_roles": "",
        ...     "guard_expr": "python:here.guard_handler('store')",
        ... }

        >>> wf = get_workflow(SAMPLE_WORKFLOW)
        >>> transition = wf.transitions.get("store")
        >>> update_transition(transition,
        ...                   title="Store",
        ...                   description="The action to store the sample",
        ...                   action="Store sample",
        ...                   new_state="stored",
        ...                   guard=guard)

    :param transition: Workflow transition definition object
    :type transition: Products.DCWorkflow.Transitions.TransitionDefinition
    :param title: (optional) the title of the transition
    :type title: string
    :param description: (optional) the descrioption of the transition
    :type title: string
    :param new_state: (optional) the state of the object after the transition
    :type new_state: string
    :param after_script: (optional) Script (Python) to run after the transition
    :type after_script: string
    :param action: (optional) the action name to display in the actions box
    :type action: string
    :param action_url: (optional) the url to use for this action. The
        %(content_url) wildcard is replaced by absolute path at runtime. If
        empty, the default `content_status_modify?workflow_action=<action_id>`
        will be used at runtime
    :type action_url: string
    :param guard: (optional) a dict with Guard properties. The supported
        properties are: 'guard_roles', 'guard_groups', 'guard_expr' and
        'guard_permissions'
    :type guard: dict
    """
    # attrs conversions
    mapping = {
        "title": "title",
        "description": "description",
        "action": "actbox_name",
        "action_url": "actbox_url",
        "after_script": "after_script_name",
        "new_state": "new_state_id",
    }
    properties = {}
    for key, property_id in mapping.items():
        default = getattr(transition, property_id)
        value = kwargs.get(key, None)
        if value is None:
            value = default
        properties[property_id] = value

    # update the transition properties
    transition.setProperties(**properties)

    # update the guard
    guard = transition.guard or Guard()
    guard_props = kwargs.get("guard")
    if guard_props:
        guard.changeFromProperties(guard_props)
    transition.guard = guard


def update_permission(state, permission_id, roles):
    """Updates the permission mappings of an existing workflow state

    :param state: Workflow state definition object
    :type state: Products.DCWorkflow.States.StateDefinition
    :param permission_id: id of the permission
    :type permission_id: string
    :param roles: List or tuple with roles to which the given permission has
        to be granted for the workflow status. If a tuple, acquire is set to
        False and roles overwritten. Thus, an empty tuple clears all roles for
        this permission and state. If a list, the existing roles are extended
        with new ones and acquire setting is not modified.
    :type roles: list/tuple
    """
    # resolve acquire
    if isinstance(roles, tuple):
        acquired = 0
    else:
        info = state.getPermissionInfo(permission_id) or {}
        acquired = info.get("acquired", 1)
        roles = set(roles)
        roles.update(info.get("roles", []))

    # sort them for human-friendly reading on retrieval
    roles = tuple(sorted(roles))

    # add this permission to the workflow if not globally defined yet
    wf = get_workflow(state)
    if permission_id not in wf.permissions:
        wf.permissions = wf.permissions + (permission_id,)

    # set the permission
    logger.info("{}.{}: '{}' (acquired={}): '{}'".format(
        wf.id, state.id, permission_id, repr(acquired), ', '.join(roles)))
    state.setPermission(permission_id, acquired, roles)


def copy_permissions(source, destination):
    """Copies the permission mappings of a workflow state to another

    :param source: Workflow state definition object used as source
    :type source: Products.DCWorkflow.States.StateDefinition
    :param destination: Workflow state definition object used as destination
    :type destination: Products.DCWorkflow.States.StateDefinition
    """
    for permission in source.permissions:
        info = source.getPermissionInfo(permission)
        roles = info.get("roles") or []
        acquired = info.get("acquired", 1)
        # update the roles for this permission at destination
        destination.setPermission(permission, acquired, sorted(roles))


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

senaite / senaite.core

Push — 2.x ( 1c5375...9d7a5b )

senaite.core.api.workflow.get_transition() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like