bika.lims.browser.workflow.RequestContextAware.get_action() - Code Metrics - Inspection of "Moved request data methods to superclass" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( e646f1...7151f0 )

by Ramon

created 2019-03-11 14:08 UTC

RequestContextAware.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)

    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 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)


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			def get_uids(self):
37			"""Returns a uids list of the objects this action must be performed
38			against to. If no values for uids param found in the request, returns
39			the uid of the current context
40			"""
41			uids = self.get_uids_from_request()
42			if not uids and api.is_object(self.context):
43			uids = [api.get_uid(self.context)]
44			return uids
45
46			def get_uids_from_request(self):
47			"""Returns a list of uids from the request
48			"""
49			uids = self.request.get("uids", "")
50			if isinstance(uids, basestring):
51			uids = uids.split(",")
52			unique_uids = collections.OrderedDict().fromkeys(uids).keys()
53			return filter(api.is_uid, unique_uids)
54
55			def get_action(self):
56			"""Returns the action to be taken from the request. Returns None if no
57			action is found
58			"""
59			action = self.request.get("workflow_action_id", None)
60			action = self.request.get("workflow_action", action)
61			if not action:
62			return None
63
64			# A condition in the form causes Plone to sometimes send two actions
65			# This usually happens when the previous action was not managed properly
66			# and the request was not able to complete, so the previous form value
67			# is kept, together with the new one.
68			if type(action) in (list, tuple):
69			actions = list(set(action))
70			if len(actions) > 0:
71			logger.warn("Multiple actions in request: {}. Fallback to '{}'"
72			.format(repr(actions), actions[-1]))
73			action = actions[-1]
74			return action
75
76
77			class WorkflowActionHandler(RequestContextAware):
78			"""Handler in charge of processing workflow action requests from views and
79			if necessary, delegates actions to third-party subscribers/adapters.
80			"""
81			def __init__(self, context, request):
82			super(WorkflowActionHandler, self).__init__(context, request)
83
84			# TODO This "context_uid" dance is probably no longer necessary
85			self.request["context_uid"] = ""
86			if api.is_object(self.context):
87			self.request["context_uid"] = api.get_uid(self.context)
88
89			def __call__(self):
90			# Get the id of the action to be performed
91			action = self.get_action()
92			if not action:
93			return self.redirect(message=_("No action defined."), level="error")
94
95			# Get the UIDs of the objects the action needs to be done against
96			uids = self.get_uids()
97			if not uids:
98			return self.redirect(message=_("No items selected."),
99			level="warning")
100
101			# Find out if there is an adapter registered for the current context
102			# able to handle the requested action
103			adapter_name = "workflow_action_{}".format(action)
104			adapter = queryMultiAdapter((self.context, self.request),
105			interface=IWorkflowActionAdapter,
106			name=adapter_name)
107			if not adapter:
108			# No adapter found, use the generic one
109			adapter = WorkflowActionGenericAdapter(self.context, self.request)
110			else:
111			adapter_module = adapter.__module__
112			adapter_class = adapter.__class__.__name__
113			logger.info("Action '{}' managed by {}.{}"
114			.format(action, adapter_module, adapter_class))
115
116			# Handle the action
117			if IWorkflowActionUIDsAdapter.providedBy(adapter):
118			return adapter(action, uids)
119
120			objects = api.search(dict(UID=uids), UID_CATALOG)
121			objects = map(api.get_object, objects)
122			return adapter(action, objects)
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 ( e646f1...7151f0 )

RequestContextAware.get_action() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like