bika.lims.browser.workflow.WorkflowActionHandler.get_action() - Code Metrics - Inspection of "Reverted changes directly pushed to master" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 7151f0...475656 )

by Ramon

created 2019-03-11 14:43 UTC

WorkflowActionHandler.get_action() A

↳ Parent: bika.lims.browser.workflow

Complexity

Conditions

Size

Total Lines	20
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	12
dl	0
loc	20
rs	9.8
c	0
b	0
f	0
cc	4
nop	1

import collections

from Products.Archetypes.config import UID_CATALOG
from bika.lims import api
from bika.lims import bikaMessageFactory as _
from bika.lims import logger
from bika.lims.interfaces import IWorkflowActionAdapter, \
    IWorkflowActionUIDsAdapter
from bika.lims.workflow import ActionHandlerPool
from bika.lims.workflow import doActionFor as do_action_for
from zope.component import queryMultiAdapter
from zope.component.interfaces import implements


class RequestContextAware(object):

    def __init__(self, context, request):
        self.context = context
        self.request = request
        self.back_url = self.context.absolute_url()

    def redirect(self, redirect_url=None, message=None, level="info"):
        """Redirect with a message
        """
        if redirect_url is None:
            redirect_url = self.back_url
        if message is not None:
            self.add_status_message(message, level)
        return self.request.response.redirect(redirect_url)

    def add_status_message(self, message, level="info"):
        """Set a portal status message
        """
        return self.context.plone_utils.addPortalMessage(message, level)


class WorkflowActionHandler(RequestContextAware):
    """Handler in charge of processing workflow action requests from views and
    if necessary, delegates actions to third-party subscribers/adapters.
    """
    def __init__(self, context, request):
        super(WorkflowActionHandler, self).__init__(context, request)

        # TODO This "context_uid" dance is probably no longer necessary
        self.request["context_uid"] = ""
        if api.is_object(self.context):
            self.request["context_uid"] = api.get_uid(self.context)

    def __call__(self):
        # Get the id of the action to be performed
        action = self.get_action()
        if not action:
            return self.redirect(message=_("No action defined."), level="error")

        # Get the UIDs of the objects the action needs to be done against
        uids = self.get_uids()
        if not uids:
            return self.redirect(message=_("No items selected."),
                                 level="warning")

        # Find out if there is an adapter registered for the current context
        # able to handle the requested action
        adapter_name = "workflow_action_{}".format(action)
        adapter = queryMultiAdapter((self.context, self.request),
                                    interface=IWorkflowActionAdapter,
                                    name=adapter_name)
        if not adapter:
            # No adapter found, use the generic one
            adapter = WorkflowActionGenericAdapter(self.context, self.request)
        else:
            adapter_module = adapter.__module__
            adapter_class = adapter.__class__.__name__
            logger.info("Action '{}' managed by {}.{}"
                        .format(action, adapter_module, adapter_class))

        # Handle the action
        if IWorkflowActionUIDsAdapter.providedBy(adapter):
            return adapter(action, uids)

        objects = api.search(dict(UID=uids), UID_CATALOG)
        objects = map(api.get_object, objects)
        return adapter(action, objects)

    def get_uids(self):
        """Returns a uids list of the objects this action must be performed
        against to. If no values for uids param found in the request, returns
        the uid of the current context
        """
        uids = self.get_uids_from_request()
        if not uids and api.is_object(self.context):
            uids = [api.get_uid(self.context)]
        return uids

    def get_uids_from_request(self):
        """Returns a list of uids from the request
        """
        uids = self.request.get("uids", "")
        if isinstance(uids, basestring):
            uids = uids.split(",")
        unique_uids = collections.OrderedDict().fromkeys(uids).keys()
        return filter(api.is_uid, unique_uids)

    def get_action(self):
        """Returns the action to be taken from the request. Returns None if no
        action is found
        """
        action = self.request.get("workflow_action_id", None)
        action = self.request.get("workflow_action", action)
        if not action:
            return None

        # A condition in the form causes Plone to sometimes send two actions
        # This usually happens when the previous action was not managed properly
        # and the request was not able to complete, so the previous form value
        # is kept, together with the new one.
        if type(action) in (list, tuple):
            actions = list(set(action))
            if len(actions) > 0:
                logger.warn("Multiple actions in request: {}. Fallback to '{}'"
                            .format(repr(actions), actions[-1]))
            action = actions[-1]
        return action


class WorkflowActionGenericAdapter(RequestContextAware):
    """General purpose adapter for processing workflow action requests
    """
    implements(IWorkflowActionAdapter)

    def __call__(self, action, objects):
        """Performs the given action to the objects passed in
        """
        transitioned = self.do_action(action, objects)
        if not transitioned:
            return self.redirect(message=_("No changes made."), level="warning")

        # TODO Here we could handle subscriber adapters here?

        # Redirect the user to success page
        return self.success(transitioned)

    def do_action(self, action, objects):
        """Performs the workflow transition passed in and returns the list of
        objects that have been successfully transitioned
        """
        transitioned = []
        ActionHandlerPool.get_instance().queue_pool()
        for obj in objects:
            obj = api.get_object(obj)
            success, message = do_action_for(obj, action)
            if success:
                transitioned.append(obj)
        ActionHandlerPool.get_instance().resume()
        return transitioned

    def is_context_only(self, objects):
        """Returns whether the action applies to the current context only
        """
        if len(objects) > 1:
            return False
        return self.context in objects

    def success(self, objects):
        """Redirects the user to success page with informative message
        """
        if self.is_context_only(objects):
            return self.redirect(message=_("Changes saved."))

        ids = map(api.get_id, objects)
        message = _("Saved items: {}").format(", ".join(ids))
        return self.redirect(message=message)

    def get_form_value(self, form_key, object_brain_uid, default=None):
        """Returns a value from the request's form for the given uid, if any
        """
        if form_key not in self.request.form:
            return default

        uid = object_brain_uid
        if not api.is_uid(uid):
            uid = api.get_uid(object_brain_uid)

        values = self.request.form.get(form_key)
        if isinstance(values, list):
            if len(values) == 0:
                return default
            if len(values) > 1:
                logger.warn("Multiple set of values for {}".format(form_key))
            values = values[0]

        return values.get(uid, default)


1			import collections
2
3			from Products.Archetypes.config import UID_CATALOG
4			from bika.lims import api
5			from bika.lims import bikaMessageFactory as _
6			from bika.lims import logger
7			from bika.lims.interfaces import IWorkflowActionAdapter, \
8			IWorkflowActionUIDsAdapter
9			from bika.lims.workflow import ActionHandlerPool
10			from bika.lims.workflow import doActionFor as do_action_for
11			from zope.component import queryMultiAdapter
12			from zope.component.interfaces import implements
13
14
15			class RequestContextAware(object):
16
17			def __init__(self, context, request):
18			self.context = context
19			self.request = request
20			self.back_url = self.context.absolute_url()
21
22			def redirect(self, redirect_url=None, message=None, level="info"):
23			"""Redirect with a message
24			"""
25			if redirect_url is None:
26			redirect_url = self.back_url
27			if message is not None:
28			self.add_status_message(message, level)
29			return self.request.response.redirect(redirect_url)
30
31			def add_status_message(self, message, level="info"):
32			"""Set a portal status message
33			"""
34			return self.context.plone_utils.addPortalMessage(message, level)
35
36
37			class WorkflowActionHandler(RequestContextAware):
38			"""Handler in charge of processing workflow action requests from views and
39			if necessary, delegates actions to third-party subscribers/adapters.
40			"""
41			def __init__(self, context, request):
42			super(WorkflowActionHandler, self).__init__(context, request)
43
44			# TODO This "context_uid" dance is probably no longer necessary
45			self.request["context_uid"] = ""
46			if api.is_object(self.context):
47			self.request["context_uid"] = api.get_uid(self.context)
48
49			def __call__(self):
50			# Get the id of the action to be performed
51			action = self.get_action()
52			if not action:
53			return self.redirect(message=_("No action defined."), level="error")
54
55			# Get the UIDs of the objects the action needs to be done against
56			uids = self.get_uids()
57			if not uids:
58			return self.redirect(message=_("No items selected."),
59			level="warning")
60
61			# Find out if there is an adapter registered for the current context
62			# able to handle the requested action
63			adapter_name = "workflow_action_{}".format(action)
64			adapter = queryMultiAdapter((self.context, self.request),
65			interface=IWorkflowActionAdapter,
66			name=adapter_name)
67			if not adapter:
68			# No adapter found, use the generic one
69			adapter = WorkflowActionGenericAdapter(self.context, self.request)
70			else:
71			adapter_module = adapter.__module__
72			adapter_class = adapter.__class__.__name__
73			logger.info("Action '{}' managed by {}.{}"
74			.format(action, adapter_module, adapter_class))
75
76			# Handle the action
77			if IWorkflowActionUIDsAdapter.providedBy(adapter):
78			return adapter(action, uids)
79
80			objects = api.search(dict(UID=uids), UID_CATALOG)
81			objects = map(api.get_object, objects)
82			return adapter(action, objects)
83
84			def get_uids(self):
85			"""Returns a uids list of the objects this action must be performed
86			against to. If no values for uids param found in the request, returns
87			the uid of the current context
88			"""
89			uids = self.get_uids_from_request()
90			if not uids and api.is_object(self.context):
91			uids = [api.get_uid(self.context)]
92			return uids
93
94			def get_uids_from_request(self):
95			"""Returns a list of uids from the request
96			"""
97			uids = self.request.get("uids", "")
98			if isinstance(uids, basestring):
99			uids = uids.split(",")
100			unique_uids = collections.OrderedDict().fromkeys(uids).keys()
101			return filter(api.is_uid, unique_uids)
102
103			def get_action(self):
104			"""Returns the action to be taken from the request. Returns None if no
105			action is found
106			"""
107			action = self.request.get("workflow_action_id", None)
108			action = self.request.get("workflow_action", action)
109			if not action:
110			return None
111
112			# A condition in the form causes Plone to sometimes send two actions
113			# This usually happens when the previous action was not managed properly
114			# and the request was not able to complete, so the previous form value
115			# is kept, together with the new one.
116			if type(action) in (list, tuple):
117			actions = list(set(action))
118			if len(actions) > 0:
119			logger.warn("Multiple actions in request: {}. Fallback to '{}'"
120			.format(repr(actions), actions[-1]))
121			action = actions[-1]
122			return action
123
124
125			class WorkflowActionGenericAdapter(RequestContextAware):
126			"""General purpose adapter for processing workflow action requests
127			"""
128			implements(IWorkflowActionAdapter)
129
130			def __call__(self, action, objects):
131			"""Performs the given action to the objects passed in
132			"""
133			transitioned = self.do_action(action, objects)
134			if not transitioned:
135			return self.redirect(message=_("No changes made."), level="warning")
136
137			# TODO Here we could handle subscriber adapters here?
138
139			# Redirect the user to success page
140			return self.success(transitioned)
141
142			def do_action(self, action, objects):
143			"""Performs the workflow transition passed in and returns the list of
144			objects that have been successfully transitioned
145			"""
146			transitioned = []
147			ActionHandlerPool.get_instance().queue_pool()
148			for obj in objects:
149			obj = api.get_object(obj)
150			success, message = do_action_for(obj, action)
151			if success:
152			transitioned.append(obj)
153			ActionHandlerPool.get_instance().resume()
154			return transitioned
155
156			def is_context_only(self, objects):
157			"""Returns whether the action applies to the current context only
158			"""
159			if len(objects) > 1:
160			return False
161			return self.context in objects
162
163			def success(self, objects):
164			"""Redirects the user to success page with informative message
165			"""
166			if self.is_context_only(objects):
167			return self.redirect(message=_("Changes saved."))
168
169			ids = map(api.get_id, objects)
170			message = _("Saved items: {}").format(", ".join(ids))
171			return self.redirect(message=message)
172
173			def get_form_value(self, form_key, object_brain_uid, default=None):
174			"""Returns a value from the request's form for the given uid, if any
175			"""
176			if form_key not in self.request.form:
177			return default
178
179			uid = object_brain_uid
180			if not api.is_uid(uid):
181			uid = api.get_uid(object_brain_uid)
182
183			values = self.request.form.get(form_key)
184			if isinstance(values, list):
185			if len(values) == 0:
186			return default
187			if len(values) > 1:
188			logger.warn("Multiple set of values for {}".format(form_key))
189			values = values[0]
190
191			return values.get(uid, default)
192

senaite / senaite.core

Push — master ( 7151f0...475656 )

WorkflowActionHandler.get_action() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like