bika.lims.browser.listing.ajax.AjaxListingView.recalculate_results() - Code Metrics - Inspection of "Fix results calculation of dependent calculations..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 6ac657...5de4b7 )

by Jordi

created 2018-12-27 10:30 UTC

AjaxListingView.recalculate_results() B

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

Complexity

Conditions

Size

Total Lines	27
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	17
dl	0
loc	27
rs	7.3333
c	0
b	0
f	0
cc	8
nop	3

# -*- 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 bika.lims.interfaces import IReferenceAnalysis
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 = set()

        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)
            # skip objects w/o transitions, e.g. retracted/rejected analyses
            if not obj_transitions:
                continue
            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 not common_tids:
                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(),
            "content_filter": self.contentFilter,
            "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,
            "fetch_transitions_on_select": self.fetch_transitions_on_select,
        }

        return config

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

        :returns: List of recalculated objects
        """
        if recalculated is None:
            recalculated = []
        # avoid double recalculation in recursion
        if obj in recalculated:
            return []
        # recalculate own result
        if obj.calculateResult(override=True):
            # append object to the list of recalculated results
            recalculated.append(obj)
        # recalculate dependent analyses
        for dep in obj.getDependents():
            if dep.calculateResult(override=True):
                # TODO the `calculateResult` method should return False here!
                if dep.getResult() in ["NA", "0/0"]:
                    continue
                recalculated.append(dep)
            # recalculate dependents of dependents
            for ddep in dep.getDependents():
                recalculated.extend(
                    self.recalculate_results(
                        ddep, recalculated=recalculated))
        return recalculated

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

    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

    @set_application_json_header
    @returns_safe_json
    @inject_runtime
    def ajax_set_fields(self):
        """Set multiple fields

        The POST Payload needs to provide the following data:

        :save_queue: A mapping of {UID: {fieldname: fieldvalue, ...}}
        """

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

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

        save_queue = payload.get("save_queue")

        updated_objects = set()
        for uid, data in save_queue.iteritems():
            # get the object
            obj = api.get_object_by_uid(uid)

            # update the fields
            for name, value in data.iteritems():
                updated_objects.update(self.set_field(obj, name, value))

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

        # UIDs of updated objects
        updated_uids = map(api.get_uid, updated_objects)

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

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

senaite / senaite.core

Push — master ( 6ac657...5de4b7 )

AjaxListingView.recalculate_results() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like