bika.lims.browser.listing.ajax.AjaxListingView.base_info() - Code Metrics - Inspection of "Analysis Requests with rejected analyses were not..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 4b0f3a...3fdd65 )

by Ramon

created 2018-11-25 21:26 UTC

AjaxListingView.base_info() A

↳ Parent: bika.lims.browser.listing.ajax

Complexity

Conditions

Size

Total Lines	11
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	8
dl	0
loc	11
rs	10
c	0
b	0
f	0
cc	1
nop	2

# -*- coding: utf-8 -*-

import inspect
import json
import urllib

from bika.lims import api
from bika.lims.browser import BrowserView
from bika.lims.browser.listing.decorators import inject_runtime
from bika.lims.browser.listing.decorators import returns_safe_json
from bika.lims.browser.listing.decorators import set_application_json_header
from bika.lims.browser.listing.decorators import translate
from bika.lims.interfaces import IRoutineAnalysis
from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile
from zope.interface import implements
from zope.publisher.interfaces import IPublishTraverse


class AjaxListingView(BrowserView):
    """Mixin Class for the ajax enabled listing table

    The main purpose of this class is to provide a JSON API endpoint for the
    ReactJS based listing table.
    """
    implements(IPublishTraverse)
    contents_table_template = ViewPageTemplateFile(
        "templates/contents_table.pt")

    def __init__(self, context, request):
        super(AjaxListingView, self).__init__(context, request)
        self.traverse_subpath = []

    def ajax_contents_table(self, *args, **kwargs):
        """Render the ReactJS enabled contents table template

        It is called from the `BikaListingView.contents_table` method
        """
        return self.contents_table_template()

    def publishTraverse(self, request, name):
        """Called before __call__ for each path name and allows to dispatch
        subpaths to methods
        """
        self.traverse_subpath.append(name)
        return self

    def handle_subpath(self, prefix="ajax_"):
        """Dispatch the subpath to a method prefixed with the given prefix

        N.B. Only the first subpath element will be dispatched to a method and
             the rest will be passed as arguments to this method
        """
        if len(self.traverse_subpath) < 1:
            return {}

        # check if the method exists
        func_arg = self.traverse_subpath[0]
        func_name = "{}{}".format(prefix, func_arg)
        func = getattr(self, func_name, None)
        if func is None:
            raise NameError("Invalid function name")

        # Additional provided path segments after the function name are handled
        # as positional arguments
        args = self.traverse_subpath[1:]

        # check mandatory arguments
        func_sig = inspect.getargspec(func)
        # positional arguments after `self` argument
        required_args = func_sig.args[1:]

        if len(args) < len(required_args):
            raise ValueError("Wrong signature, please use '{}/{}'"
                             .format(func_arg, "/".join(required_args)))
        return func(*args)

    def get_json(self, encoding="utf8"):
        """Extracts the JSON from the request
        """
        body = self.request.get("BODY", "{}")

        def encode_hook(pairs):
            """This hook is called for dicitionaries on JSON deserialization

            It is used to encode unicode strings with the given encoding,
            because ZCatalogs have sometimes issues with unicode queries.
            """
            new_pairs = []
            for key, value in pairs.iteritems():
                # Encode the key
                if isinstance(key, unicode):
                    key = key.encode(encoding)
                # Encode the value
                if isinstance(value, unicode):
                    value = value.encode(encoding)
                new_pairs.append((key, value))
            return dict(new_pairs)

        return json.loads(body, object_hook=encode_hook)

    def error(self, message, status=500, **kw):
        """Set a JSON error object and a status to the response
        """
        self.request.response.setStatus(status)
        result = {"success": False, "errors": message, "status": status}
        result.update(kw)
        return result

    @translate
    def get_columns(self):
        """Returns the `columns` dictionary of the view
        """
        return self.columns

    @translate
    def get_review_states(self):
        """Returns the `review_states` list of the view
        """
        return self.review_states

    def to_form_data(self, data):
        """Prefix all data keys with `self.form_id`

        This is needed to inject the POST Body to the request form data, so
        that the `BikaListingView.folderitems` finds them.
        """
        out = {}
        for key, value in data.iteritems():
            if not key.startswith(self.form_id):
                k = "{}_{}".format(self.form_id, key)
            out[k] = value
        return out

    def get_api_url(self):
        """Calculate the API URL of this view
        """
        url = self.context.absolute_url()
        view_name = self.__name__
        return "{}/{}".format(url, view_name)

    def get_transitions_for(self, obj):
        """Get the allowed transitions for the given object
        """
        return api.get_transitions_for(obj)

    def get_allowed_transitions_for(self, uids):
        """Retrieves all transitions from the given UIDs and calculate the
        ones which have all in common (intersection).
        """

        # Handle empty list of objects
        if not uids:
            return []

        # allowed transitions
        transitions = []

        # allowed transitions of the current workflow
        allowed_transitions = self.review_state.get("transitions", [])
        allowed_transition_ids = map(
            lambda t: t.get("id"), allowed_transitions)

        # internal mapping of transition id -> transition
        transitions_by_tid = {}

        # get the custom transitions of the current review_state
        custom_transitions = self.review_state.get("custom_transitions", [])

        # N.B. we use set() here to handle dupes in the views gracefully
        custom_tids = set()
        for transition in custom_transitions:
            tid = transition.get("id")
            custom_tids.add(tid)
            transitions_by_tid[tid] = transition

        # transition ids all objects have in common
        common_tids = None

        for uid in uids:
            # TODO: Research how to avoid the object wakeup here
            obj = api.get_object_by_uid(uid)
            obj_transitions = self.get_transitions_for(obj)
            tids = []
            for transition in obj_transitions:
                tid = transition.get("id")
                if allowed_transition_ids:
                    if tid not in allowed_transition_ids:
                        continue
                    tids.append(tid)
                else:
                    # no restrictions by the selected review_state
                    tids.append(tid)
                transitions_by_tid[tid] = transition

            if common_tids is None:
                common_tids = set(tids)

            common_tids = common_tids.intersection(tids)

        # union the common and custom transitions
        all_transition_ids = common_tids.union(custom_tids)

        def sort_transitions(a, b):
            transition_weights = {
                "invalidate": 100,
                "retract": 90,
                "reject": 90,
                "cancel": 80,
                "deactivate": 70,
                "unassign": 70,
                "publish": 60,
                "republish": 50,
                "prepublish": 50,
                "partition": 40,
                "assign": 30,
                "receive": 20,
                "submit": 10,
            }
            w1 = transition_weights.get(a, 0)
            w2 = transition_weights.get(b, 0)
            return cmp(w1, w2)

        for tid in sorted(all_transition_ids, cmp=sort_transitions):
            transition = transitions_by_tid.get(tid)
            transitions.append(transition)

        return transitions

    def get_category_uid(self, brain_or_object, accessor="getCategoryUID"):
        """Get the category UID from the brain or object

        This will be used to speed up the listing by categories
        """
        attr = getattr(brain_or_object, accessor, None)
        if attr is None:
            return ""
        if callable(attr):
            return attr()
        return attr

    @returns_safe_json
    def ajax_review_states(self):
        """Returns the `review_states` list of the view as JSON
        """
        return self.get_review_states()

    @returns_safe_json
    def ajax_columns(self):
        """Returns the `columns` dictionary of the view as JSON
        """
        return self.get_columns()

    @translate
    def get_folderitems(self):
        """This method calls the folderitems method
        """
        # workaround for `pagesize` handling in BikaListing
        pagesize = self.get_pagesize()
        self.pagesize = pagesize

        # get the folderitems
        self.update()
        self.before_render()

        return self.folderitems()

    def get_selected_uids(self, folderitems, uids_to_keep=None):
        """Lookup selected UIDs from the folderitems
        """
        selected_uids = []
        if uids_to_keep:
            selected_uids = uids_to_keep

        for folderitem in folderitems:
            uid = folderitem.get("uid")
            if uid in selected_uids:
                continue
            if folderitem.get("selected", False):
                selected_uids.append(folderitem.get("uid"))
        return selected_uids

    def get_listing_config(self):
        """Get the configuration settings of the current listing view
        """

        config = {
            "form_id": self.form_id,
            "review_states": self.get_review_states(),
            "columns": self.get_columns(),
            "allow_edit": self.allow_edit,
            "api_url": self.get_api_url(),
            "catalog": self.catalog,
            "catalog_indexes": self.get_catalog_indexes(),
            "categories": self.categories,
            "expand_all_categories": self.expand_all_categories,
            "limit_from": self.limit_from,
            "pagesize": self.pagesize,
            "post_action": self.getPOSTAction(),
            "review_state": self.review_state.get("id", "default"),
            "select_checkbox_name": self.select_checkbox_name,
            "show_categories": self.show_categories,
            "show_column_toggles": self.show_column_toggles,
            "show_more": self.show_more,
            "show_select_all_checkbox": self.show_select_all_checkbox,
            "show_select_column": self.show_select_column,
            "show_table_footer": self.show_table_footer,
            "show_workflow_action_buttons": self.show_workflow_action_buttons,
            "sort_on": self.get_sort_on(),
            "sort_order": self.get_sort_order(),
            "show_search": self.show_search,
        }

        return config

    def recalculate_results(self, obj):
        """Recalculate the result of the object and its dependents

        :returns: List of recalculated objects
        """
        recalculated_objects = []
        # recalculate own result
        obj.calculateResult(override=True, cascade=True)
        # recalculate dependent analyses
        for dep in obj.getDependents():
            if dep.calculateResult(override=True, cascade=True):
                recalculated_objects.append(dep)
        return recalculated_objects

    def is_analysis(self, obj):
        """Check if the object is an analysis
        """
        return IRoutineAnalysis.providedBy(obj)

    def lookup_schema_field(self, obj, fieldname):
        """Lookup  a schema field by name

        :returns: Schema field or None
        """
        # Lookup the field by the given name
        field = obj.getField(fieldname)
        if field is None:
            # strip "get" from the fieldname
            if fieldname.startswith("get"):
                fieldname = fieldname.lstrip("get")
                field = obj.get(fieldname)
        return field

    def set_field(self, obj, name, value):
        """Set the value

        :returns: List of updated/changed objects
        """

        # set of updated objects
        updated_objects = []

        # lookup the schema field
        field = self.lookup_schema_field(obj, name)

        # field exists, set it with the value
        if field:
            obj.edit(**{field.getName(): value})
            updated_objects.append(obj)

        # check if the object is an analysis and has an interim
        if self.is_analysis(obj):
            interims = obj.getInterimFields()
            interim_keys = map(lambda i: i.get("keyword"), interims)
            if name in interim_keys:
                for interim in interims:
                    if interim.get("keyword") == name:
                        interim["value"] = value
                # set the new interim fields
                obj.setInterimFields(interims)
            # recalculate dependent results for result and interim fields
            if name == "Result" or name in interim_keys:
                updated_objects.append(obj)
                updated_objects.extend(self.recalculate_results(obj))

        # unify the list of updated objects
        updated_objects = list(set(updated_objects))

        # reindex updated objects
        map(lambda o: o.reindexObject(), updated_objects)

        return updated_objects

    def get_children_hook(self, parent_uid, child_uids=None):
        """Custom hook to be implemented by subclass
        """
        raise NotImplementedError("Must be implemented by subclass")

    @set_application_json_header
    @returns_safe_json
    @inject_runtime
    def ajax_folderitems(self):
        """Calls the `folderitems` method of the view and returns it as JSON

        1. Extract the HTTP POST payload from the request
        2. Convert the payload to HTTP form data and inject it to the request
        3. Call the `folderitems` method
        4. Prepare a data structure for the ReactJS listing app
        """

        # Get the HTTP POST JSON Payload
        payload = self.get_json()

        # Fake a HTTP GET request with parameters, so that the `bika_listing`
        # view handles them correctly.
        form_data = self.to_form_data(payload)

        # this serves `request.form.get` calls
        self.request.form.update(form_data)

        # this serves `request.get` calls
        self.request.other.update(form_data)

        # generate a query string from the form data
        query_string = urllib.urlencode(form_data)

        # get the folder items
        folderitems = self.get_folderitems()

        # Process selected UIDs and their allowed transitions
        uids_to_keep = payload.get("selected_uids")
        selected_uids = self.get_selected_uids(folderitems, uids_to_keep)
        transitions = self.get_allowed_transitions_for(selected_uids)

        # get the view config
        config = self.get_listing_config()

        # prepare the response object
        data = {
            "count": len(folderitems),
            "folderitems": folderitems,
            "query_string": query_string,
            "selected_uids": selected_uids,
            "total": self.total,
            "transitions": transitions,
        }

        # update the config
        data.update(config)

        # XXX fix broken `sort_on` lookup in BikaListing
        sort_on = payload.get("sort_on")
        if sort_on in self.get_catalog_indexes():
            data["sort_on"] = sort_on

        return data

    @set_application_json_header
    @returns_safe_json
    @inject_runtime
    def ajax_transitions(self):
        """Returns a list of possible transitions
        """
        # Get the HTTP POST JSON Payload
        payload = self.get_json()

        # Get the selected UIDs
        uids = payload.get("selected_uids", [])

        # ----------------------------------8<---------------------------------
        # XXX Temporary (cut out as soon as possible)
        #
        # Some listings inject custom transitions before rendering the
        # folderitems, e.g. the worksheets folder listing view.
        # This can be removed as soon as all the relevant permission checks are
        # done on the object only and not by manual role checking in the view.

        # Fake a HTTP GET request with parameters, so that the `bika_listing`
        # view handles them correctly.
        form_data = self.to_form_data(payload)

        # this serves `request.form.get` calls
        self.request.form.update(form_data)

        # this serves `request.get` calls
        self.request.other.update(form_data)

        # Call the update and before_render hook, because these might modify
        # the allowed and custom transitions (and columns probably as well)
        self.update()
        self.before_render()
        # ----------------------------------8<---------------------------------

        # get the allowed transitions
        transitions = self.get_allowed_transitions_for(uids)

        # prepare the response object
        data = {
            "transitions": transitions,
        }

        return data

    @set_application_json_header
    @returns_safe_json
    @inject_runtime
    def ajax_get_children(self):
        """Get the children of a folderitem

        Required POST JSON Payload:

        :uid: UID of the parent folderitem
        """

        # Get the HTTP POST JSON Payload
        payload = self.get_json()

        # extract the parent UID
        parent_uid = payload.get("parent_uid")
        child_uids = payload.get("child_uids", [])

        try:
            children = self.get_children_hook(
                parent_uid, child_uids=child_uids)
        except NotImplementedError:
            # lookup by catalog search
            self.contentFilter = {"UID": child_uids}
            children = self.get_folderitems()
            # ensure the children have a reference to the parent
            for child in children:
                child["parent"] = parent_uid

        # prepare the response object
        data = {
            "children": children
        }

        return data

    @set_application_json_header
    @returns_safe_json
    @inject_runtime
    def ajax_query_folderitems(self):
        """Get folderitems with a catalog query

        Required POST JSON Payload:

        :query: Catalog query to use
        :type query: dictionary
        """

        # Get the HTTP POST JSON Payload
        payload = self.get_json()

        # extract the catalog query
        query = payload.get("query", {})

        valid_catalog_indexes = self.get_catalog_indexes()

        # sanity check
        for key, value in query.iteritems():
            if key not in valid_catalog_indexes:
                return self.error("{} is not a valid catalog index".format(key))

        # set the content filter
        self.contentFilter = query

        # get the folderitems
        folderitems = self.get_folderitems()

        # prepare the response object
        data = {
            "count": len(folderitems),
            "folderitems": folderitems,
        }

        return data

    @set_application_json_header
    @returns_safe_json
    @inject_runtime
    def ajax_set(self):
        """Set a value of an editable field

        The POST Payload needs to provide the following data:

        :uid: UID of the object changed
        :name: Column name as provided by the self.columns key
        :value: The value to save
        :item: The folderitem containing the data
        """

        # Get the HTTP POST JSON Payload
        payload = self.get_json()

        required = ["uid", "name", "value"]
        if not all(map(lambda k: k in payload, required)):
            return self.error("Payload needs to provide the keys {}"
                              .format(", ".join(required)), status=400)

        uid = payload.get("uid")
        name = payload.get("name")
        value = payload.get("value")

        # get the object
        obj = api.get_object_by_uid(uid)

        # set the field
        updated_objects = self.set_field(obj, name, value)

        if not updated_objects:
            return self.error("Failed to set field '{}'".format(name), 500)

        # get the folderitems
        self.contentFilter["UID"] = map(api.get_uid, updated_objects)
        folderitems = self.get_folderitems()

        # prepare the response object
        data = {
            "count": len(folderitems),
            "folderitems": folderitems,
        }

        return data


1			# -- coding: utf-8 --
2
3			import inspect
4			import json
5			import urllib
6
7			from bika.lims import api
8			from bika.lims.browser import BrowserView
9			from bika.lims.browser.listing.decorators import inject_runtime
10			from bika.lims.browser.listing.decorators import returns_safe_json
11			from bika.lims.browser.listing.decorators import set_application_json_header
12			from bika.lims.browser.listing.decorators import translate
13			from bika.lims.interfaces import IRoutineAnalysis
14			from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile
15			from zope.interface import implements
16			from zope.publisher.interfaces import IPublishTraverse
17
18
19			class AjaxListingView(BrowserView):
20			"""Mixin Class for the ajax enabled listing table
21
22			The main purpose of this class is to provide a JSON API endpoint for the
23			ReactJS based listing table.
24			"""
25			implements(IPublishTraverse)
26			contents_table_template = ViewPageTemplateFile(
27			"templates/contents_table.pt")
28
29			def __init__(self, context, request):
30			super(AjaxListingView, self).__init__(context, request)
31			self.traverse_subpath = []
32
33			def ajax_contents_table(self, args, *kwargs):
34			"""Render the ReactJS enabled contents table template
35
36			It is called from the `BikaListingView.contents_table` method
37			"""
38			return self.contents_table_template()
39
40			def publishTraverse(self, request, name):
41			"""Called before __call__ for each path name and allows to dispatch
42			subpaths to methods
43			"""
44			self.traverse_subpath.append(name)
45			return self
46
47			def handle_subpath(self, prefix="ajax_"):
48			"""Dispatch the subpath to a method prefixed with the given prefix
49
50			N.B. Only the first subpath element will be dispatched to a method and
51			the rest will be passed as arguments to this method
52			"""
53			if len(self.traverse_subpath) < 1:
54			return {}
55
56			# check if the method exists
57			func_arg = self.traverse_subpath[0]
58			func_name = "{}{}".format(prefix, func_arg)
59			func = getattr(self, func_name, None)
60			if func is None:
61			raise NameError("Invalid function name")
62
63			# Additional provided path segments after the function name are handled
64			# as positional arguments
65			args = self.traverse_subpath[1:]
66
67			# check mandatory arguments
68			func_sig = inspect.getargspec(func)
69			# positional arguments after `self` argument
70			required_args = func_sig.args[1:]
71
72			if len(args) < len(required_args):
73			raise ValueError("Wrong signature, please use '{}/{}'"
74			.format(func_arg, "/".join(required_args)))
75			return func(*args)
76
77			def get_json(self, encoding="utf8"):
78			"""Extracts the JSON from the request
79			"""
80			body = self.request.get("BODY", "{}")
81
82			def encode_hook(pairs):
83			"""This hook is called for dicitionaries on JSON deserialization
84
85			It is used to encode unicode strings with the given encoding,
86			because ZCatalogs have sometimes issues with unicode queries.
87			"""
88			new_pairs = []
89			for key, value in pairs.iteritems():
90			# Encode the key
91			if isinstance(key, unicode):
92			key = key.encode(encoding)
93			# Encode the value
94			if isinstance(value, unicode):
95			value = value.encode(encoding)
96			new_pairs.append((key, value))
97			return dict(new_pairs)
98
99			return json.loads(body, object_hook=encode_hook)
100
101			def error(self, message, status=500, **kw):
102			"""Set a JSON error object and a status to the response
103			"""
104			self.request.response.setStatus(status)
105			result = {"success": False, "errors": message, "status": status}
106			result.update(kw)
107			return result
108
109			@translate
110			def get_columns(self):
111			"""Returns the `columns` dictionary of the view
112			"""
113			return self.columns
114
115			@translate
116			def get_review_states(self):
117			"""Returns the `review_states` list of the view
118			"""
119			return self.review_states
120
121			def to_form_data(self, data):
122			"""Prefix all data keys with `self.form_id`
123
124			This is needed to inject the POST Body to the request form data, so
125			that the `BikaListingView.folderitems` finds them.
126			"""
127			out = {}
128			for key, value in data.iteritems():
129			if not key.startswith(self.form_id):
130			k = "{}_{}".format(self.form_id, key)
131			out[k] = value
132			return out
133
134			def get_api_url(self):
135			"""Calculate the API URL of this view
136			"""
137			url = self.context.absolute_url()
138			view_name = self.__name__
139			return "{}/{}".format(url, view_name)
140
141			def get_transitions_for(self, obj):
142			"""Get the allowed transitions for the given object
143			"""
144			return api.get_transitions_for(obj)
145
146			def get_allowed_transitions_for(self, uids):
147			"""Retrieves all transitions from the given UIDs and calculate the
148			ones which have all in common (intersection).
149			"""
150
151			# Handle empty list of objects
152			if not uids:
153			return []
154
155			# allowed transitions
156			transitions = []
157
158			# allowed transitions of the current workflow
159			allowed_transitions = self.review_state.get("transitions", [])
160			allowed_transition_ids = map(
161			lambda t: t.get("id"), allowed_transitions)
162
163			# internal mapping of transition id -> transition
164			transitions_by_tid = {}
165
166			# get the custom transitions of the current review_state
167			custom_transitions = self.review_state.get("custom_transitions", [])
168
169			# N.B. we use set() here to handle dupes in the views gracefully
170			custom_tids = set()
171			for transition in custom_transitions:
172			tid = transition.get("id")
173			custom_tids.add(tid)
174			transitions_by_tid[tid] = transition
175
176			# transition ids all objects have in common
177			common_tids = None
178
179			for uid in uids:
180			# TODO: Research how to avoid the object wakeup here
181			obj = api.get_object_by_uid(uid)
182			obj_transitions = self.get_transitions_for(obj)
183			tids = []
184			for transition in obj_transitions:
185			tid = transition.get("id")
186			if allowed_transition_ids:
187			if tid not in allowed_transition_ids:
188			continue
189			tids.append(tid)
190			else:
191			# no restrictions by the selected review_state
192			tids.append(tid)
193			transitions_by_tid[tid] = transition
194
195			if common_tids is None:
196			common_tids = set(tids)
197
198			common_tids = common_tids.intersection(tids)
199
200			# union the common and custom transitions
201			all_transition_ids = common_tids.union(custom_tids)
202
203			def sort_transitions(a, b):
204			transition_weights = {
205			"invalidate": 100,
206			"retract": 90,
207			"reject": 90,
208			"cancel": 80,
209			"deactivate": 70,
210			"unassign": 70,
211			"publish": 60,
212			"republish": 50,
213			"prepublish": 50,
214			"partition": 40,
215			"assign": 30,
216			"receive": 20,
217			"submit": 10,
218			}
219			w1 = transition_weights.get(a, 0)
220			w2 = transition_weights.get(b, 0)
221			return cmp(w1, w2)
222
223			for tid in sorted(all_transition_ids, cmp=sort_transitions):
224			transition = transitions_by_tid.get(tid)
225			transitions.append(transition)
226
227			return transitions
228
229			def get_category_uid(self, brain_or_object, accessor="getCategoryUID"):
230			"""Get the category UID from the brain or object
231
232			This will be used to speed up the listing by categories
233			"""
234			attr = getattr(brain_or_object, accessor, None)
235			if attr is None:
236			return ""
237			if callable(attr):
238			return attr()
239			return attr
240
241			@returns_safe_json
242			def ajax_review_states(self):
243			"""Returns the `review_states` list of the view as JSON
244			"""
245			return self.get_review_states()
246
247			@returns_safe_json
248			def ajax_columns(self):
249			"""Returns the `columns` dictionary of the view as JSON
250			"""
251			return self.get_columns()
252
253			@translate
254			def get_folderitems(self):
255			"""This method calls the folderitems method
256			"""
257			# workaround for `pagesize` handling in BikaListing
258			pagesize = self.get_pagesize()
259			self.pagesize = pagesize
260
261			# get the folderitems
262			self.update()
263			self.before_render()
264
265			return self.folderitems()
266
267			def get_selected_uids(self, folderitems, uids_to_keep=None):
268			"""Lookup selected UIDs from the folderitems
269			"""
270			selected_uids = []
271			if uids_to_keep:
272			selected_uids = uids_to_keep
273
274			for folderitem in folderitems:
275			uid = folderitem.get("uid")
276			if uid in selected_uids:
277			continue
278			if folderitem.get("selected", False):
279			selected_uids.append(folderitem.get("uid"))
280			return selected_uids
281
282			def get_listing_config(self):
283			"""Get the configuration settings of the current listing view
284			"""
285
286			config = {
287			"form_id": self.form_id,
288			"review_states": self.get_review_states(),
289			"columns": self.get_columns(),
290			"allow_edit": self.allow_edit,
291			"api_url": self.get_api_url(),
292			"catalog": self.catalog,
293			"catalog_indexes": self.get_catalog_indexes(),
294			"categories": self.categories,
295			"expand_all_categories": self.expand_all_categories,
296			"limit_from": self.limit_from,
297			"pagesize": self.pagesize,
298			"post_action": self.getPOSTAction(),
299			"review_state": self.review_state.get("id", "default"),
300			"select_checkbox_name": self.select_checkbox_name,
301			"show_categories": self.show_categories,
302			"show_column_toggles": self.show_column_toggles,
303			"show_more": self.show_more,
304			"show_select_all_checkbox": self.show_select_all_checkbox,
305			"show_select_column": self.show_select_column,
306			"show_table_footer": self.show_table_footer,
307			"show_workflow_action_buttons": self.show_workflow_action_buttons,
308			"sort_on": self.get_sort_on(),
309			"sort_order": self.get_sort_order(),
310			"show_search": self.show_search,
311			}
312
313			return config
314
315			def recalculate_results(self, obj):
316			"""Recalculate the result of the object and its dependents
317
318			:returns: List of recalculated objects
319			"""
320			recalculated_objects = []
321			# recalculate own result
322			obj.calculateResult(override=True, cascade=True)
323			# recalculate dependent analyses
324			for dep in obj.getDependents():
325			if dep.calculateResult(override=True, cascade=True):
326			recalculated_objects.append(dep)
327			return recalculated_objects
328
329			def is_analysis(self, obj):
330			"""Check if the object is an analysis
331			"""
332			return IRoutineAnalysis.providedBy(obj)
333
334			def lookup_schema_field(self, obj, fieldname):
335			"""Lookup a schema field by name
336
337			:returns: Schema field or None
338			"""
339			# Lookup the field by the given name
340			field = obj.getField(fieldname)
341			if field is None:
342			# strip "get" from the fieldname
343			if fieldname.startswith("get"):
344			fieldname = fieldname.lstrip("get")
345			field = obj.get(fieldname)
346			return field
347
348			def set_field(self, obj, name, value):
349			"""Set the value
350
351			:returns: List of updated/changed objects
352			"""
353
354			# set of updated objects
355			updated_objects = []
356
357			# lookup the schema field
358			field = self.lookup_schema_field(obj, name)
359
360			# field exists, set it with the value
361			if field:
362			obj.edit(**{field.getName(): value})
363			updated_objects.append(obj)
364
365			# check if the object is an analysis and has an interim
366			if self.is_analysis(obj):
367			interims = obj.getInterimFields()
368			interim_keys = map(lambda i: i.get("keyword"), interims)
369			if name in interim_keys:
370			for interim in interims:
371			if interim.get("keyword") == name:
372			interim["value"] = value
373			# set the new interim fields
374			obj.setInterimFields(interims)
375			# recalculate dependent results for result and interim fields
376			if name == "Result" or name in interim_keys:
377			updated_objects.append(obj)
378			updated_objects.extend(self.recalculate_results(obj))
379
380			# unify the list of updated objects
381			updated_objects = list(set(updated_objects))
382
383			# reindex updated objects
384			map(lambda o: o.reindexObject(), updated_objects)
385
386			return updated_objects
387
388			def get_children_hook(self, parent_uid, child_uids=None):
389			"""Custom hook to be implemented by subclass
390			"""
391			raise NotImplementedError("Must be implemented by subclass")
392
393			@set_application_json_header
394			@returns_safe_json
395			@inject_runtime
396			def ajax_folderitems(self):
397			"""Calls the `folderitems` method of the view and returns it as JSON
398
399			1. Extract the HTTP POST payload from the request
400			2. Convert the payload to HTTP form data and inject it to the request
401			3. Call the `folderitems` method
402			4. Prepare a data structure for the ReactJS listing app
403			"""
404
405			# Get the HTTP POST JSON Payload
406			payload = self.get_json()
407
408			# Fake a HTTP GET request with parameters, so that the `bika_listing`
409			# view handles them correctly.
410			form_data = self.to_form_data(payload)
411
412			# this serves `request.form.get` calls
413			self.request.form.update(form_data)
414
415			# this serves `request.get` calls
416			self.request.other.update(form_data)
417
418			# generate a query string from the form data
419			query_string = urllib.urlencode(form_data)
420
421			# get the folder items
422			folderitems = self.get_folderitems()
423
424			# Process selected UIDs and their allowed transitions
425			uids_to_keep = payload.get("selected_uids")
426			selected_uids = self.get_selected_uids(folderitems, uids_to_keep)
427			transitions = self.get_allowed_transitions_for(selected_uids)
428
429			# get the view config
430			config = self.get_listing_config()
431
432			# prepare the response object
433			data = {
434			"count": len(folderitems),
435			"folderitems": folderitems,
436			"query_string": query_string,
437			"selected_uids": selected_uids,
438			"total": self.total,
439			"transitions": transitions,
440			}
441
442			# update the config
443			data.update(config)
444
445			# XXX fix broken `sort_on` lookup in BikaListing
446			sort_on = payload.get("sort_on")
447			if sort_on in self.get_catalog_indexes():
448			data["sort_on"] = sort_on
449
450			return data
451
452			@set_application_json_header
453			@returns_safe_json
454			@inject_runtime
455			def ajax_transitions(self):
456			"""Returns a list of possible transitions
457			"""
458			# Get the HTTP POST JSON Payload
459			payload = self.get_json()
460
461			# Get the selected UIDs
462			uids = payload.get("selected_uids", [])
463
464			# ----------------------------------8<---------------------------------
465			# XXX Temporary (cut out as soon as possible)
466			#
467			# Some listings inject custom transitions before rendering the
468			# folderitems, e.g. the worksheets folder listing view.
469			# This can be removed as soon as all the relevant permission checks are
470			# done on the object only and not by manual role checking in the view.
471
472			# Fake a HTTP GET request with parameters, so that the `bika_listing`
473			# view handles them correctly.
474			form_data = self.to_form_data(payload)
475
476			# this serves `request.form.get` calls
477			self.request.form.update(form_data)
478
479			# this serves `request.get` calls
480			self.request.other.update(form_data)
481
482			# Call the update and before_render hook, because these might modify
483			# the allowed and custom transitions (and columns probably as well)
484			self.update()
485			self.before_render()
486			# ----------------------------------8<---------------------------------
487
488			# get the allowed transitions
489			transitions = self.get_allowed_transitions_for(uids)
490
491			# prepare the response object
492			data = {
493			"transitions": transitions,
494			}
495
496			return data
497
498			@set_application_json_header
499			@returns_safe_json
500			@inject_runtime
501			def ajax_get_children(self):
502			"""Get the children of a folderitem
503
504			Required POST JSON Payload:
505
506			:uid: UID of the parent folderitem
507			"""
508
509			# Get the HTTP POST JSON Payload
510			payload = self.get_json()
511
512			# extract the parent UID
513			parent_uid = payload.get("parent_uid")
514			child_uids = payload.get("child_uids", [])
515
516			try:
517			children = self.get_children_hook(
518			parent_uid, child_uids=child_uids)
519			except NotImplementedError:
520			# lookup by catalog search
521			self.contentFilter = {"UID": child_uids}
522			children = self.get_folderitems()
523			# ensure the children have a reference to the parent
524			for child in children:
525			child["parent"] = parent_uid
526
527			# prepare the response object
528			data = {
529			"children": children
530			}
531
532			return data
533
534			@set_application_json_header
535			@returns_safe_json
536			@inject_runtime
537			def ajax_query_folderitems(self):
538			"""Get folderitems with a catalog query
539
540			Required POST JSON Payload:
541
542			:query: Catalog query to use
543			:type query: dictionary
544			"""
545
546			# Get the HTTP POST JSON Payload
547			payload = self.get_json()
548
549			# extract the catalog query
550			query = payload.get("query", {})
551
552			valid_catalog_indexes = self.get_catalog_indexes()
553
554			# sanity check
555			for key, value in query.iteritems():
556			if key not in valid_catalog_indexes:
557			return self.error("{} is not a valid catalog index".format(key))
558
559			# set the content filter
560			self.contentFilter = query
561
562			# get the folderitems
563			folderitems = self.get_folderitems()
564
565			# prepare the response object
566			data = {
567			"count": len(folderitems),
568			"folderitems": folderitems,
569			}
570
571			return data
572
573			@set_application_json_header
574			@returns_safe_json
575			@inject_runtime
576			def ajax_set(self):
577			"""Set a value of an editable field
578
579			The POST Payload needs to provide the following data:
580
581			:uid: UID of the object changed
582			:name: Column name as provided by the self.columns key
583			:value: The value to save
584			:item: The folderitem containing the data
585			"""
586
587			# Get the HTTP POST JSON Payload
588			payload = self.get_json()
589
590			required = ["uid", "name", "value"]
591			if not all(map(lambda k: k in payload, required)):
592			return self.error("Payload needs to provide the keys {}"
593			.format(", ".join(required)), status=400)
594
595			uid = payload.get("uid")
596			name = payload.get("name")
597			value = payload.get("value")
598
599			# get the object
600			obj = api.get_object_by_uid(uid)
601
602			# set the field
603			updated_objects = self.set_field(obj, name, value)
604
605			if not updated_objects:
606			return self.error("Failed to set field '{}'".format(name), 500)
607
608			# get the folderitems
609			self.contentFilter["UID"] = map(api.get_uid, updated_objects)
610			folderitems = self.get_folderitems()
611
612			# prepare the response object
613			data = {
614			"count": len(folderitems),
615			"folderitems": folderitems,
616			}
617
618			return data
619

senaite / senaite.core

Push — master ( 4b0f3a...3fdd65 )

AjaxListingView.base_info() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like