bika.lims.workflow.wasTransitionPerformed() - Code Metrics - Inspection of "Disable caching of persistent objects in action po..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 7bdfa8...29d318 )

by Ramon

created 2019-03-11 10:53 UTC

bika.lims.workflow.wasTransitionPerformed() A

↳ Parent: bika.lims.workflow

Complexity

Conditions

Size

Total Lines	6
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	3
dl	0
loc	6
rs	10
c	0
b	0
f	0
cc	1
nop	2

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE
#
# Copyright 2018 by it's authors.
# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.

import collections
import sys

from AccessControl.SecurityInfo import ModuleSecurityInfo
from Products.CMFCore.WorkflowCore import WorkflowException
from Products.CMFCore.utils import getToolByName
from bika.lims import PMF
from bika.lims import api
from bika.lims import logger
from bika.lims.browser import ulocalized_time
from bika.lims.interfaces import IJSONReadExtender
from bika.lims.jsonapi import get_include_fields
from bika.lims.utils import changeWorkflowState
from bika.lims.utils import t
from bika.lims.workflow.indexes import ACTIONS_TO_INDEXES
from zope.interface import implements

security = ModuleSecurityInfo('bika.lims.workflow')
security.declarePublic('guard_handler')

_marker = object()

def skip(instance, action, peek=False, unskip=False):
    """Returns True if the transition is to be SKIPPED

        peek - True just checks the value, does not set.
        unskip - remove skip key (for manual overrides).

    called with only (instance, action_id), this will set the request variable preventing the
    cascade's from re-transitioning the object and return None.
    """

    uid = callable(instance.UID) and instance.UID() or instance.UID
    skipkey = "%s_%s" % (uid, action)
    if 'workflow_skiplist' not in instance.REQUEST:
        if not peek and not unskip:
            instance.REQUEST['workflow_skiplist'] = [skipkey, ]
    else:
        if skipkey in instance.REQUEST['workflow_skiplist']:
            if unskip:
                instance.REQUEST['workflow_skiplist'].remove(skipkey)
            else:
                return True
        else:
            if not peek and not unskip:
                instance.REQUEST["workflow_skiplist"].append(skipkey)


def doActionFor(instance, action_id, idxs=None):
    """Tries to perform the transition to the instance.
    Object is reindexed after the transition takes place, but only if succeeds.
    If idxs is set, only these indexes will be reindexed. Otherwise, will try
    to use the indexes defined in ACTIONS_TO_INDEX mapping if any.
    :param instance: Object to be transitioned
    :param action_id: transition id
    :param idxs: indexes to be reindexed after the transition
    :returns: True if the transition has been performed, together with message
    :rtype: tuple (bool,str)
    """
    if not instance:
        return False, ""

    if isinstance(instance, list):
        # TODO Workflow . Check if this is strictly necessary
        # This check is here because sometimes Plone creates a list
        # from submitted form elements.
        logger.warn("Got a list of obj in doActionFor!")
        if len(instance) > 1:
            logger.warn(
                "doActionFor is getting an instance parameter which is a list "
                "with more than one item. Instance: '{}', action_id: '{}'"
                .format(instance, action_id)
            )

        return doActionFor(instance=instance[0], action_id=action_id, idxs=idxs)

    # Since a given transition can cascade or promote to other objects, we want
    # to reindex all objects for which the transition succeed at once, at the
    # end of process. Otherwise, same object will be reindexed multiple times
    # unnecessarily. Also, ActionsHandlerPool ensures the same transition is not
    # applied twice to the same object due to cascade/promote recursions.
    pool = ActionHandlerPool.get_instance()
    if pool.succeed(instance, action_id):
        return False, "Transition {} for {} already done"\
             .format(action_id, instance.getId())

    # Return False if transition is not permitted
    if not isTransitionAllowed(instance, action_id):
        return False, "Transition {} for {} is not allowed"\
            .format(action_id, instance.getId())

    # Add this batch process to the queue
    pool.queue_pool()
    succeed = False
    message = ""
    workflow = getToolByName(instance, "portal_workflow")
    try:
        workflow.doActionFor(instance, action_id)
        succeed = True
    except WorkflowException as e:
        message = str(e)
        curr_state = getCurrentState(instance)
        clazz_name = instance.__class__.__name__
        logger.warning(
            "Transition '{0}' not allowed: {1} '{2}' ({3})"\
            .format(action_id, clazz_name, instance.getId(), curr_state))
        logger.error(message)

    # If no indexes to reindex have been defined, try to use those defined in
    # the ACTIONS_TO_INDEXES mapping. Reindexing only those indexes that might
    # be affected by the transition boosts the overall performance!.
    if idxs is None:
        portal_type = instance.portal_type
        idxs = ACTIONS_TO_INDEXES.get(portal_type, {}).get(action_id, [])

    # Add the current object to the pool and resume
    pool.push(instance, action_id, succeed, idxs=idxs)
    pool.resume()

    return succeed, message


def call_workflow_event(instance, event, after=True):
    """Calls the instance's workflow event
    """
    if not event.transition:
        return False

    portal_type = instance.portal_type
    wf_module = _load_wf_module('{}.events'.format(portal_type.lower()))
    if not wf_module:
        return False

    # Inspect if event_<transition_id> function exists in the module
    prefix = after and "after" or "before"
    func_name = "{}_{}".format(prefix, event.transition.id)
    func = getattr(wf_module, func_name, False)
    if not func:
        return False

    logger.info('WF event: {0}.events.{1}'
                .format(portal_type.lower(), func_name))
    func(instance)
    return True


def BeforeTransitionEventHandler(instance, event):
    """ This event is executed before each transition and delegates further
    actions to 'workflow.<portal_type>.events.before_<transition_id> function
    if exists for the instance passed in.
    :param instance: the instance to be transitioned
    :type instance: ATContentType
    :param event: event that holds the transition to be performed
    :type event: IObjectEvent
    """
    call_workflow_event(instance, event, after=False)


def AfterTransitionEventHandler(instance, event):
    """ This event is executed after each transition and delegates further
    actions to 'workflow.<portal_type>.events.after_<transition_id> function
    if exists for the instance passed in.
    :param instance: the instance that has been transitioned
    :type instance: ATContentType
    :param event: event that holds the transition performed
    :type event: IObjectEvent
    """
    if call_workflow_event(instance, event, after=True):
        return

    # Try with old AfterTransitionHandler dance...
    # TODO CODE TO BE REMOVED AFTER PORTING workflow_script_*/*_transition_event
    if not event.transition:
        return
    # Set the request variable preventing cascade's from re-transitioning.
    if skip(instance, event.transition.id):
        return
    # Because at this point, the object has been transitioned already, but
    # further actions are probably needed still, so be sure is reindexed
    # before going forward.
    instance.reindexObject()
    key = 'after_{0}_transition_event'.format(event.transition.id)
    after_event = getattr(instance, key, False)
    if not after_event:
        # TODO Workflow. this conditional is only for backwards compatibility,
        # to be removed when all workflow_script_* methods in contents are
        # replaced by the more explicity signature 'after_*_transition_event'
        key = 'workflow_script_' + event.transition.id
        after_event = getattr(instance, key, False)
    if not after_event:
        return
    after_event()


def get_workflow_actions(obj):
    """ Compile a list of possible workflow transitions for this object
    """

    def translate(id):
        return t(PMF(id + "_transition_title"))

    transids = getAllowedTransitions(obj)
    actions = [{'id': it, 'title': translate(it)} for it in transids]
    return actions


def isTransitionAllowed(instance, transition_id):
    """Checks if the object can perform the transition passed in.
    :returns: True if transition can be performed
    :rtype: bool
    """
    wf_tool = getToolByName(instance, "portal_workflow")
    for wf_id in wf_tool.getChainFor(instance):
        wf = wf_tool.getWorkflowById(wf_id)
        if wf and wf.isActionSupported(instance, transition_id):
            return True

    return False


def getAllowedTransitions(instance):
    """Returns a list with the transition ids that can be performed against
    the instance passed in.
    :param instance: A content object
    :type instance: ATContentType
    :returns: A list of transition/action ids
    :rtype: list
    """
    wftool = getToolByName(instance, "portal_workflow")
    transitions = wftool.getTransitionsFor(instance)
    return [trans['id'] for trans in transitions]


def wasTransitionPerformed(instance, transition_id):
    """Checks if the transition has already been performed to the object
    Instance's workflow history is checked.
    """
    transitions = getReviewHistoryActionsList(instance)
    return transition_id in transitions


def getReviewHistoryActionsList(instance, reverse=False):
    """Returns a list with the actions performed to the instance
    """
    review_history = getReviewHistory(instance, reverse=reverse)
    return map(lambda event: event["action"], review_history)


def get_prev_status_from_history(instance, status=None):
    """Returns the previous status of the object. If status is set, returns the
    previous status before the object reached the status passed in.
    If instance has reached the status passed in more than once, only the last
    one is considered.
    """
    target = status or api.get_workflow_status_of(instance)
    history = getReviewHistory(instance, reverse=True)
    history = map(lambda event: event["review_state"], history)
    if target not in history or history.index(target) == len(history)-1:
        return None
    return history[history.index(target)+1]


def getReviewHistory(instance, reverse=True):
    """Returns the review history for the instance
    :returns: the list of historic events as dicts
    """
    review_history = []
    workflow = getToolByName(instance, 'portal_workflow')
    try:
        review_history = list(workflow.getInfoFor(instance, 'review_history'))
    except WorkflowException:
        logger.error("Unable to retrieve review history from {}:{}"
                     .format(instance.portal_type, instance.getId()))
    if reverse:
        # invert the list, so we always see the most recent matching event
        review_history.reverse()
    return review_history

def getCurrentState(obj, stateflowid='review_state'):
    """ The current state of the object for the state flow id specified
        Return empty if there's no workflow state for the object and flow id
    """
    return api.get_workflow_status_of(obj, stateflowid)

def in_state(obj, states, stateflowid='review_state'):
    """ Returns if the object passed matches with the states passed in
    """
    if not states:
        return False
    obj_state = getCurrentState(obj, stateflowid=stateflowid)
    return obj_state in states

def getTransitionActor(obj, action_id):
    """Returns the actor that performed a given transition. If transition has
    not been perormed, or current user has no privileges, returns None
    :return: the username of the user that performed the transition passed-in
    :type: string
    """
    review_history = getReviewHistory(obj)
    for event in review_history:
        if event.get('action') == action_id:
            return event.get('actor')
    return None


def getTransitionDate(obj, action_id, return_as_datetime=False):
    """
    Returns date of action for object. Sometimes we need this date in Datetime
    format and that's why added return_as_datetime param.
    """
    review_history = getReviewHistory(obj)
    for event in review_history:
        if event.get('action') == action_id:
            evtime = event.get('time')
            if return_as_datetime:
                return evtime
            if evtime:
                value = ulocalized_time(evtime, long_format=True,
                                        time_only=False, context=obj)
                return value
    return None


def getTransitionUsers(obj, action_id, last_user=False):
    """
    This function returns a list with the users who have done the transition.
    :action_id: a sring as the transition id.
    :last_user: a boolean to return only the last user triggering the
        transition or all of them.
    :returns: a list of user ids.
    """
    workflow = getToolByName(obj, 'portal_workflow')
    users = []
    try:
        # https://jira.bikalabs.com/browse/LIMS-2242:
        # Sometimes the workflow history is inexplicably missing!
        review_history = list(workflow.getInfoFor(obj, 'review_history'))
    except WorkflowException:
        logger.error(
            "workflow history is inexplicably missing."
            " https://jira.bikalabs.com/browse/LIMS-2242")
        return users
    # invert the list, so we always see the most recent matching event
    review_history.reverse()
    for event in review_history:
        if event.get('action', '') == action_id:
            value = event.get('actor', '')
            users.append(value)
            if last_user:
                return users
    return users


def guard_handler(instance, transition_id):
    """Generic workflow guard handler that returns true if the transition_id
    passed in can be performed to the instance passed in.

    This function is called automatically by a Script (Python) located at
    bika/lims/skins/guard_handler.py, which in turn is fired by Zope when an
    expression like "python:here.guard_handler('<transition_id>')" is set to
    any given guard (used by default in all bika's DC Workflow guards).

    Walks through bika.lims.workflow.<obj_type>.guards and looks for a function
    that matches with 'guard_<transition_id>'. If found, calls the function and
    returns its value (true or false). If not found, returns True by default.

    :param instance: the object for which the transition_id has to be evaluated
    :param transition_id: the id of the transition
    :type instance: ATContentType
    :type transition_id: string
    :return: true if the transition can be performed to the passed in instance
    :rtype: bool
    """
    if not instance:
        return True
    clazz_name = instance.portal_type
    # Inspect if bika.lims.workflow.<clazzname>.<guards> module exists
    wf_module = _load_wf_module('{0}.guards'.format(clazz_name.lower()))
    if not wf_module:
        return True

    # Inspect if guard_<transition_id> function exists in the above module
    key = 'guard_{0}'.format(transition_id)
    guard = getattr(wf_module, key, False)
    if not guard:
        return True

    #logger.info('{0}.guards.{1}'.format(clazz_name.lower(), key))
    return guard(instance)


def _load_wf_module(module_relative_name):
    """Loads a python module based on the module relative name passed in.

    At first, tries to get the module from sys.modules. If not found there, the
    function tries to load it by using importlib. Returns None if no module
    found or importlib is unable to load it because of errors.
    Eg:
        _load_wf_module('sample.events')

    will try to load the module 'bika.lims.workflow.sample.events'

    :param modrelname: relative name of the module to be loaded
    :type modrelname: string
    :return: the module
    :rtype: module
    """
    if not module_relative_name:
        return None
    if not isinstance(module_relative_name, basestring):
        return None

    rootmodname = __name__
    modulekey = '{0}.{1}'.format(rootmodname, module_relative_name)
    if modulekey in sys.modules:
        return sys.modules.get(modulekey, None)

    # Try to load the module recursively
    modname = None
    tokens = module_relative_name.split('.')
    for part in tokens:
        modname = '.'.join([modname, part]) if modname else part
        import importlib
        try:
            _module = importlib.import_module('.'+modname, package=rootmodname)
            if not _module:
                return None
        except Exception:
            return None
    return sys.modules.get(modulekey, None)


class JSONReadExtender(object):

    """- Adds the list of possible transitions to each object, if 'transitions'
    is specified in the include_fields.
    """

    implements(IJSONReadExtender)

    def __init__(self, context):
        self.context = context

    def __call__(self, request, data):
        include_fields = get_include_fields(request)
        if not include_fields or "transitions" in include_fields:
            data['transitions'] = get_workflow_actions(self.context)


# TODO Workflow - ActionsPool - Better use ActionHandlerPool
class ActionsPool(object):
    """Handles transitions of multiple objects at once
    """
    def __init__(self):
        self.actions_pool = collections.OrderedDict()

    def add(self, instance, action_id):
        uid = api.get_uid(instance)
        self.actions_pool[uid] = {"instance": instance,
                                  "action_id": action_id}

    def resume(self):
        action_handler = ActionHandlerPool.get_instance()
        action_handler.queue_pool()
        outcome = collections.OrderedDict()
        for uid, values in self.actions_pool.items():
            instance = values["instance"]
            action_id = values["action_id"]
            outcome[uid] = doActionFor(instance, action_id)
        self.actions_pool = collections.OrderedDict()
        action_handler.resume()
        return outcome


class ActionHandlerPool(object):
    """Singleton to handle concurrent transitions
    """
    __instance = None

    @staticmethod
    def get_instance():
        """Returns the current instance of ActionHandlerPool
        """
        if ActionHandlerPool.__instance == None:
            ActionHandlerPool()
        return ActionHandlerPool.__instance

    def __init__(self):
        if ActionHandlerPool.__instance != None:
            raise Exception("Use ActionHandlerPool.get_instance()")
        self.objects = collections.OrderedDict()
        self.num_calls = 0
        ActionHandlerPool.__instance = self

    def __len__(self):
        """Number of objects in the pool
        """
        return len(self.objects)

    def queue_pool(self):
        """Notifies that a new batch of jobs is about to begin
        """
        self.num_calls += 1

    def push(self, instance, action, success, idxs=_marker):
        """Adds an instance into the pool, to be reindexed on resume
        """
        uid = api.get_uid(instance)
        info = self.objects.get(uid, {})
        idx = [] if idxs is _marker else idxs
        info[action] = {'success': success, 'idxs': idx}
        self.objects[uid] = info

    def succeed(self, instance, action):
        """Returns if the task for the instance took place successfully
        """
        uid = api.get_uid(instance)
        return self.objects.get(uid, {}).get(action, {}).get('success', False)

    def resume(self):
        """Resumes the pool and reindex all objects processed
        """
        self.num_calls -= 1
        if self.num_calls > 0:
            return
        logger.info("Resume actions for {} objects".format(len(self)))
        processed = list()
        for uid, info in self.objects.items():
            if uid in processed:
                continue

            idxs = self.get_indexes(uid)
            idxs_str = idxs and ', '.join(idxs) or "-- All indexes --"
            instance = api.get_object_by_uid(uid, default=None)
            if instance:
                logger.info(
                    "Reindexing {}: {}".format(instance.getId(), idxs_str))
                instance.reindexObject(idxs=self.get_indexes(uid))
            processed.append(uid)
        logger.info("Objects processed: {}".format(len(processed)))
        self.objects = collections.OrderedDict()

    def get_indexes(self, uid):
        """Returns the names of the indexes to be reindexed for the object with
        the uid passed in. If no indexes for this object have been specified
        within the action pool job, returns an empty list (reindex all).
        Otherwise, return all the indexes that have been specified for the
        object within the action pool job.
        """
        idxs = []
        info = self.objects.get(uid, {})
        for action_id, value in info.items():
            obj_idxs = value.get('idxs', None)
            if obj_idxs is None:
                # Don't reindex!
                continue
            elif len(obj_idxs) == 0:
                # Reindex all indexes!
                return []
            idxs.extend(obj_idxs)
        # Always reindex review_state and is_active
        idxs.extend(["review_state", "is_active"])
        return list(set(idxs))


def push_reindex_to_actions_pool(obj, idxs=None):
    """Push a reindex job to the actions handler pool
    """
    indexes = idxs and idxs or []
    pool = ActionHandlerPool.get_instance()
    pool.push(obj, "reindex", success=True, idxs=indexes)


1			# -- coding: utf-8 --
2			#
3			# This file is part of SENAITE.CORE
4			#
5			# Copyright 2018 by it's authors.
6			# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.
7
8			import collections
9			import sys
10
11			from AccessControl.SecurityInfo import ModuleSecurityInfo
12			from Products.CMFCore.WorkflowCore import WorkflowException
13			from Products.CMFCore.utils import getToolByName
14			from bika.lims import PMF
15			from bika.lims import api
16			from bika.lims import logger
17			from bika.lims.browser import ulocalized_time
18			from bika.lims.interfaces import IJSONReadExtender
19			from bika.lims.jsonapi import get_include_fields
20			from bika.lims.utils import changeWorkflowState
21			from bika.lims.utils import t
22			from bika.lims.workflow.indexes import ACTIONS_TO_INDEXES
23			from zope.interface import implements
24
25			security = ModuleSecurityInfo('bika.lims.workflow')
26			security.declarePublic('guard_handler')
27
28			_marker = object()
29
30			def skip(instance, action, peek=False, unskip=False):
31			"""Returns True if the transition is to be SKIPPED
32
33			peek - True just checks the value, does not set.
34			unskip - remove skip key (for manual overrides).
35
36			called with only (instance, action_id), this will set the request variable preventing the
37			cascade's from re-transitioning the object and return None.
38			"""
39
40			uid = callable(instance.UID) and instance.UID() or instance.UID
41			skipkey = "%s_%s" % (uid, action)
42			if 'workflow_skiplist' not in instance.REQUEST:
43			if not peek and not unskip:
44			instance.REQUEST['workflow_skiplist'] = [skipkey, ]
45			else:
46			if skipkey in instance.REQUEST['workflow_skiplist']:
47			if unskip:
48			instance.REQUEST['workflow_skiplist'].remove(skipkey)
49			else:
50			return True
51			else:
52			if not peek and not unskip:
53			instance.REQUEST["workflow_skiplist"].append(skipkey)
54
55
56			def doActionFor(instance, action_id, idxs=None):
57			"""Tries to perform the transition to the instance.
58			Object is reindexed after the transition takes place, but only if succeeds.
59			If idxs is set, only these indexes will be reindexed. Otherwise, will try
60			to use the indexes defined in ACTIONS_TO_INDEX mapping if any.
61			:param instance: Object to be transitioned
62			:param action_id: transition id
63			:param idxs: indexes to be reindexed after the transition
64			:returns: True if the transition has been performed, together with message
65			:rtype: tuple (bool,str)
66			"""
67			if not instance:
68			return False, ""
69
70			if isinstance(instance, list):
71			# TODO Workflow . Check if this is strictly necessary
72			# This check is here because sometimes Plone creates a list
73			# from submitted form elements.
74			logger.warn("Got a list of obj in doActionFor!")
75			if len(instance) > 1:
76			logger.warn(
77			"doActionFor is getting an instance parameter which is a list "
78			"with more than one item. Instance: '{}', action_id: '{}'"
79			.format(instance, action_id)
80			)
81
82			return doActionFor(instance=instance[0], action_id=action_id, idxs=idxs)
83
84			# Since a given transition can cascade or promote to other objects, we want
85			# to reindex all objects for which the transition succeed at once, at the
86			# end of process. Otherwise, same object will be reindexed multiple times
87			# unnecessarily. Also, ActionsHandlerPool ensures the same transition is not
88			# applied twice to the same object due to cascade/promote recursions.
89			pool = ActionHandlerPool.get_instance()
90			if pool.succeed(instance, action_id):
91			return False, "Transition {} for {} already done"\
92			.format(action_id, instance.getId())
93
94			# Return False if transition is not permitted
95			if not isTransitionAllowed(instance, action_id):
96			return False, "Transition {} for {} is not allowed"\
97			.format(action_id, instance.getId())
98
99			# Add this batch process to the queue
100			pool.queue_pool()
101			succeed = False
102			message = ""
103			workflow = getToolByName(instance, "portal_workflow")
104			try:
105			workflow.doActionFor(instance, action_id)
106			succeed = True
107			except WorkflowException as e:
108			message = str(e)
109			curr_state = getCurrentState(instance)
110			clazz_name = instance.__class__.__name__
111			logger.warning(
112			"Transition '{0}' not allowed: {1} '{2}' ({3})"\
113			.format(action_id, clazz_name, instance.getId(), curr_state))
114			logger.error(message)
115
116			# If no indexes to reindex have been defined, try to use those defined in
117			# the ACTIONS_TO_INDEXES mapping. Reindexing only those indexes that might
118			# be affected by the transition boosts the overall performance!.
119			if idxs is None:
120			portal_type = instance.portal_type
121			idxs = ACTIONS_TO_INDEXES.get(portal_type, {}).get(action_id, [])
122
123			# Add the current object to the pool and resume
124			pool.push(instance, action_id, succeed, idxs=idxs)
125			pool.resume()
126
127			return succeed, message
128
129
130			def call_workflow_event(instance, event, after=True):
131			"""Calls the instance's workflow event
132			"""
133			if not event.transition:
134			return False
135
136			portal_type = instance.portal_type
137			wf_module = _load_wf_module('{}.events'.format(portal_type.lower()))
138			if not wf_module:
139			return False
140
141			# Inspect if event_<transition_id> function exists in the module
142			prefix = after and "after" or "before"
143			func_name = "{}_{}".format(prefix, event.transition.id)
144			func = getattr(wf_module, func_name, False)
145			if not func:
146			return False
147
148			logger.info('WF event: {0}.events.{1}'
149			.format(portal_type.lower(), func_name))
150			func(instance)
151			return True
152
153
154			def BeforeTransitionEventHandler(instance, event):
155			""" This event is executed before each transition and delegates further
156			actions to 'workflow.<portal_type>.events.before_<transition_id> function
157			if exists for the instance passed in.
158			:param instance: the instance to be transitioned
159			:type instance: ATContentType
160			:param event: event that holds the transition to be performed
161			:type event: IObjectEvent
162			"""
163			call_workflow_event(instance, event, after=False)
164
165
166			def AfterTransitionEventHandler(instance, event):
167			""" This event is executed after each transition and delegates further
168			actions to 'workflow.<portal_type>.events.after_<transition_id> function
169			if exists for the instance passed in.
170			:param instance: the instance that has been transitioned
171			:type instance: ATContentType
172			:param event: event that holds the transition performed
173			:type event: IObjectEvent
174			"""
175			if call_workflow_event(instance, event, after=True):
176			return
177
178			# Try with old AfterTransitionHandler dance...
179			# TODO CODE TO BE REMOVED AFTER PORTING workflow_script_/_transition_event
180			if not event.transition:
181			return
182			# Set the request variable preventing cascade's from re-transitioning.
183			if skip(instance, event.transition.id):
184			return
185			# Because at this point, the object has been transitioned already, but
186			# further actions are probably needed still, so be sure is reindexed
187			# before going forward.
188			instance.reindexObject()
189			key = 'after_{0}_transition_event'.format(event.transition.id)
190			after_event = getattr(instance, key, False)
191			if not after_event:
192			# TODO Workflow. this conditional is only for backwards compatibility,
193			# to be removed when all workflow_script_* methods in contents are
194			# replaced by the more explicity signature 'after_*_transition_event'
195			key = 'workflow_script_' + event.transition.id
196			after_event = getattr(instance, key, False)
197			if not after_event:
198			return
199			after_event()
200
201
202			def get_workflow_actions(obj):
203			""" Compile a list of possible workflow transitions for this object
204			"""
205
206			def translate(id):
207			return t(PMF(id + "_transition_title"))
208
209			transids = getAllowedTransitions(obj)
210			actions = [{'id': it, 'title': translate(it)} for it in transids]
211			return actions
212
213
214			def isTransitionAllowed(instance, transition_id):
215			"""Checks if the object can perform the transition passed in.
216			:returns: True if transition can be performed
217			:rtype: bool
218			"""
219			wf_tool = getToolByName(instance, "portal_workflow")
220			for wf_id in wf_tool.getChainFor(instance):
221			wf = wf_tool.getWorkflowById(wf_id)
222			if wf and wf.isActionSupported(instance, transition_id):
223			return True
224
225			return False
226
227
228			def getAllowedTransitions(instance):
229			"""Returns a list with the transition ids that can be performed against
230			the instance passed in.
231			:param instance: A content object
232			:type instance: ATContentType
233			:returns: A list of transition/action ids
234			:rtype: list
235			"""
236			wftool = getToolByName(instance, "portal_workflow")
237			transitions = wftool.getTransitionsFor(instance)
238			return [trans['id'] for trans in transitions]
239
240
241			def wasTransitionPerformed(instance, transition_id):
242			"""Checks if the transition has already been performed to the object
243			Instance's workflow history is checked.
244			"""
245			transitions = getReviewHistoryActionsList(instance)
246			return transition_id in transitions
247
248
249			def getReviewHistoryActionsList(instance, reverse=False):
250			"""Returns a list with the actions performed to the instance
251			"""
252			review_history = getReviewHistory(instance, reverse=reverse)
253			return map(lambda event: event["action"], review_history)
254
255
256			def get_prev_status_from_history(instance, status=None):
257			"""Returns the previous status of the object. If status is set, returns the
258			previous status before the object reached the status passed in.
259			If instance has reached the status passed in more than once, only the last
260			one is considered.
261			"""
262			target = status or api.get_workflow_status_of(instance)
263			history = getReviewHistory(instance, reverse=True)
264			history = map(lambda event: event["review_state"], history)
265			if target not in history or history.index(target) == len(history)-1:
266			return None
267			return history[history.index(target)+1]
268
269
270			def getReviewHistory(instance, reverse=True):
271			"""Returns the review history for the instance
272			:returns: the list of historic events as dicts
273			"""
274			review_history = []
275			workflow = getToolByName(instance, 'portal_workflow')
276			try:
277			review_history = list(workflow.getInfoFor(instance, 'review_history'))
278			except WorkflowException:
279			logger.error("Unable to retrieve review history from {}:{}"
280			.format(instance.portal_type, instance.getId()))
281			if reverse:
282			# invert the list, so we always see the most recent matching event
283			review_history.reverse()
284			return review_history
285
286			def getCurrentState(obj, stateflowid='review_state'):
287			""" The current state of the object for the state flow id specified
288			Return empty if there's no workflow state for the object and flow id
289			"""
290			return api.get_workflow_status_of(obj, stateflowid)
291
292			def in_state(obj, states, stateflowid='review_state'):
293			""" Returns if the object passed matches with the states passed in
294			"""
295			if not states:
296			return False
297			obj_state = getCurrentState(obj, stateflowid=stateflowid)
298			return obj_state in states
299
300			def getTransitionActor(obj, action_id):
301			"""Returns the actor that performed a given transition. If transition has
302			not been perormed, or current user has no privileges, returns None
303			:return: the username of the user that performed the transition passed-in
304			:type: string
305			"""
306			review_history = getReviewHistory(obj)
307			for event in review_history:
308			if event.get('action') == action_id:
309			return event.get('actor')
310			return None
311
312
313			def getTransitionDate(obj, action_id, return_as_datetime=False):
314			"""
315			Returns date of action for object. Sometimes we need this date in Datetime
316			format and that's why added return_as_datetime param.
317			"""
318			review_history = getReviewHistory(obj)
319			for event in review_history:
320			if event.get('action') == action_id:
321			evtime = event.get('time')
322			if return_as_datetime:
323			return evtime
324			if evtime:
325			value = ulocalized_time(evtime, long_format=True,
326			time_only=False, context=obj)
327			return value
328			return None
329
330
331			def getTransitionUsers(obj, action_id, last_user=False):
332			"""
333			This function returns a list with the users who have done the transition.
334			:action_id: a sring as the transition id.
335			:last_user: a boolean to return only the last user triggering the
336			transition or all of them.
337			:returns: a list of user ids.
338			"""
339			workflow = getToolByName(obj, 'portal_workflow')
340			users = []
341			try:
342			# https://jira.bikalabs.com/browse/LIMS-2242:
343			# Sometimes the workflow history is inexplicably missing!
344			review_history = list(workflow.getInfoFor(obj, 'review_history'))
345			except WorkflowException:
346			logger.error(
347			"workflow history is inexplicably missing."
348			" https://jira.bikalabs.com/browse/LIMS-2242")
349			return users
350			# invert the list, so we always see the most recent matching event
351			review_history.reverse()
352			for event in review_history:
353			if event.get('action', '') == action_id:
354			value = event.get('actor', '')
355			users.append(value)
356			if last_user:
357			return users
358			return users
359
360
361			def guard_handler(instance, transition_id):
362			"""Generic workflow guard handler that returns true if the transition_id
363			passed in can be performed to the instance passed in.
364
365			This function is called automatically by a Script (Python) located at
366			bika/lims/skins/guard_handler.py, which in turn is fired by Zope when an
367			expression like "python:here.guard_handler('<transition_id>')" is set to
368			any given guard (used by default in all bika's DC Workflow guards).
369
370			Walks through bika.lims.workflow.<obj_type>.guards and looks for a function
371			that matches with 'guard_<transition_id>'. If found, calls the function and
372			returns its value (true or false). If not found, returns True by default.
373
374			:param instance: the object for which the transition_id has to be evaluated
375			:param transition_id: the id of the transition
376			:type instance: ATContentType
377			:type transition_id: string
378			:return: true if the transition can be performed to the passed in instance
379			:rtype: bool
380			"""
381			if not instance:
382			return True
383			clazz_name = instance.portal_type
384			# Inspect if bika.lims.workflow.<clazzname>.<guards> module exists
385			wf_module = _load_wf_module('{0}.guards'.format(clazz_name.lower()))
386			if not wf_module:
387			return True
388
389			# Inspect if guard_<transition_id> function exists in the above module
390			key = 'guard_{0}'.format(transition_id)
391			guard = getattr(wf_module, key, False)
392			if not guard:
393			return True
394
395			#logger.info('{0}.guards.{1}'.format(clazz_name.lower(), key))
396			return guard(instance)
397
398
399			def _load_wf_module(module_relative_name):
400			"""Loads a python module based on the module relative name passed in.
401
402			At first, tries to get the module from sys.modules. If not found there, the
403			function tries to load it by using importlib. Returns None if no module
404			found or importlib is unable to load it because of errors.
405			Eg:
406			_load_wf_module('sample.events')
407
408			will try to load the module 'bika.lims.workflow.sample.events'
409
410			:param modrelname: relative name of the module to be loaded
411			:type modrelname: string
412			:return: the module
413			:rtype: module
414			"""
415			if not module_relative_name:
416			return None
417			if not isinstance(module_relative_name, basestring):
418			return None
419
420			rootmodname = __name__
421			modulekey = '{0}.{1}'.format(rootmodname, module_relative_name)
422			if modulekey in sys.modules:
423			return sys.modules.get(modulekey, None)
424
425			# Try to load the module recursively
426			modname = None
427			tokens = module_relative_name.split('.')
428			for part in tokens:
429			modname = '.'.join([modname, part]) if modname else part
430			import importlib
431			try:
432			_module = importlib.import_module('.'+modname, package=rootmodname)
433			if not _module:
434			return None
435			except Exception:
436			return None
437			return sys.modules.get(modulekey, None)
438
439
440			class JSONReadExtender(object):
441
442			"""- Adds the list of possible transitions to each object, if 'transitions'
443			is specified in the include_fields.
444			"""
445
446			implements(IJSONReadExtender)
447
448			def __init__(self, context):
449			self.context = context
450
451			def __call__(self, request, data):
452			include_fields = get_include_fields(request)
453			if not include_fields or "transitions" in include_fields:
454			data['transitions'] = get_workflow_actions(self.context)
455
456
457			# TODO Workflow - ActionsPool - Better use ActionHandlerPool
458			class ActionsPool(object):
459			"""Handles transitions of multiple objects at once
460			"""
461			def __init__(self):
462			self.actions_pool = collections.OrderedDict()
463
464			def add(self, instance, action_id):
465			uid = api.get_uid(instance)
466			self.actions_pool[uid] = {"instance": instance,
467			"action_id": action_id}
468
469			def resume(self):
470			action_handler = ActionHandlerPool.get_instance()
471			action_handler.queue_pool()
472			outcome = collections.OrderedDict()
473			for uid, values in self.actions_pool.items():
474			instance = values["instance"]
475			action_id = values["action_id"]
476			outcome[uid] = doActionFor(instance, action_id)
477			self.actions_pool = collections.OrderedDict()
478			action_handler.resume()
479			return outcome
480
481
482			class ActionHandlerPool(object):
483			"""Singleton to handle concurrent transitions
484			"""
485			__instance = None
486
487			@staticmethod
488			def get_instance():
489			"""Returns the current instance of ActionHandlerPool
490			"""
491			if ActionHandlerPool.__instance == None:
492			ActionHandlerPool()
493			return ActionHandlerPool.__instance
494
495			def __init__(self):
496			if ActionHandlerPool.__instance != None:
497			raise Exception("Use ActionHandlerPool.get_instance()")
498			self.objects = collections.OrderedDict()
499			self.num_calls = 0
500			ActionHandlerPool.__instance = self
501
502			def __len__(self):
503			"""Number of objects in the pool
504			"""
505			return len(self.objects)
506
507			def queue_pool(self):
508			"""Notifies that a new batch of jobs is about to begin
509			"""
510			self.num_calls += 1
511
512			def push(self, instance, action, success, idxs=_marker):
513			"""Adds an instance into the pool, to be reindexed on resume
514			"""
515			uid = api.get_uid(instance)
516			info = self.objects.get(uid, {})
517			idx = [] if idxs is _marker else idxs
518			info[action] = {'success': success, 'idxs': idx}
519			self.objects[uid] = info
520
521			def succeed(self, instance, action):
522			"""Returns if the task for the instance took place successfully
523			"""
524			uid = api.get_uid(instance)
525			return self.objects.get(uid, {}).get(action, {}).get('success', False)
526
527			def resume(self):
528			"""Resumes the pool and reindex all objects processed
529			"""
530			self.num_calls -= 1
531			if self.num_calls > 0:
532			return
533			logger.info("Resume actions for {} objects".format(len(self)))
534			processed = list()
535			for uid, info in self.objects.items():
536			if uid in processed:
537			continue
538
539			idxs = self.get_indexes(uid)
540			idxs_str = idxs and ', '.join(idxs) or "-- All indexes --"
541			instance = api.get_object_by_uid(uid, default=None)
542			if instance:
543			logger.info(
544			"Reindexing {}: {}".format(instance.getId(), idxs_str))
545			instance.reindexObject(idxs=self.get_indexes(uid))
546			processed.append(uid)
547			logger.info("Objects processed: {}".format(len(processed)))
548			self.objects = collections.OrderedDict()
549
550			def get_indexes(self, uid):
551			"""Returns the names of the indexes to be reindexed for the object with
552			the uid passed in. If no indexes for this object have been specified
553			within the action pool job, returns an empty list (reindex all).
554			Otherwise, return all the indexes that have been specified for the
555			object within the action pool job.
556			"""
557			idxs = []
558			info = self.objects.get(uid, {})
559			for action_id, value in info.items():
560			obj_idxs = value.get('idxs', None)
561			if obj_idxs is None:
562			# Don't reindex!
563			continue
564			elif len(obj_idxs) == 0:
565			# Reindex all indexes!
566			return []
567			idxs.extend(obj_idxs)
568			# Always reindex review_state and is_active
569			idxs.extend(["review_state", "is_active"])
570			return list(set(idxs))
571
572
573			def push_reindex_to_actions_pool(obj, idxs=None):
574			"""Push a reindex job to the actions handler pool
575			"""
576			indexes = idxs and idxs or []
577			pool = ActionHandlerPool.get_instance()
578			pool.push(obj, "reindex", success=True, idxs=indexes)
579

senaite / senaite.core

Push — master ( 7bdfa8...29d318 )

bika.lims.workflow.wasTransitionPerformed() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like