bika.lims.browser.partition_magic.PartitionMagicView.get_containers() - Code Metrics - Inspection of "Move create_partitions to utils package (#1248)" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( ac1405...5f2185 )

by Ramon

created 2019-02-20 12:40 UTC

PartitionMagicView.get_containers() A

↳ Parent: bika.lims.browser.partition_magic

Complexity

Conditions

Size

Total Lines	9
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	7
dl	0
loc	9
rs	10
c	0
b	0
f	0
cc	1
nop	1

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

from collections import OrderedDict
from collections import defaultdict

from Products.Five.browser import BrowserView
from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile
from bika.lims import api
from bika.lims import bikaMessageFactory as _
from bika.lims import logger
from bika.lims.decorators import returns_super_model
from bika.lims.utils.analysisrequest import create_partition

DEFAULT_NUMBER_OF_PARTITIONS = 0


class PartitionMagicView(BrowserView):
    """Manage Partitions of primary ARs
    """
    template = ViewPageTemplateFile("templates/partition_magic.pt")

    def __init__(self, context, request):
        super(PartitionMagicView, self).__init__(context, request)
        self.context = context
        self.request = request
        self.back_url = self.context.absolute_url()
        self.analyses_to_remove = dict()


    def __call__(self):
        form = self.request.form

        # Form submit toggle
        form_submitted = form.get("submitted", False)

        # Buttons
        form_preview = form.get("button_preview", False)
        form_create = form.get("button_create", False)
        form_cancel = form.get("button_cancel", False)

        objs = self.get_objects()

        # No ARs selected
        if not objs:
            return self.redirect(message=_("No items selected"),
                                 level="warning")

        # Handle preview
        if form_submitted and form_preview:
            logger.info("*** PREVIEW ***")

        # Handle create
        if form_submitted and form_create:
            logger.info("*** CREATE PARTITIONS ***")

            partitions = []

            # create the partitions
            for partition in form.get("partitions", []):
                primary_uid = partition.get("primary_uid")
                sampletype_uid = partition.get("sampletype_uid")
                container_uid = partition.get("container_uid")
                preservation_uid = partition.get("preservation_uid")
                if not primary_uid:
                    continue

                # The creation of partitions w/o analyses is allowed. Maybe the
                # user wants to add the analyses later manually or wants to keep
                # this partition stored in a freezer for some time
                # Note we set "remove_primary_analyses" to False cause we want
                # user to be able to add same analyses to different partitions.
                analyses_uids = partition.get("analyses", [])
                partition = create_partition(
                    request=self.request,
                    analysis_request=primary_uid,
                    sample_type=sampletype_uid,
                    container=container_uid,
                    preservation=preservation_uid,
                    analyses=analyses_uids,
                    remove_primary_analyses=False,
                )
                partitions.append(partition)

                logger.info("Successfully created partition: {}".format(
                    api.get_path(partition)))

            if not partitions:
                # If no partitions were created, show a warning message
                return self.redirect(message=_("No partitions were created"))

            # Remove analyses from primary Analysis Requests
            self.remove_primary_analyses()

            message = _("Created {} partitions: {}".format(
                len(partitions), ", ".join(map(api.get_title, partitions))))
            return self.redirect(message=message)

        # Handle cancel
        if form_submitted and form_cancel:
            logger.info("*** CANCEL ***")
            return self.redirect(message=_("Partitioning canceled"))

        return self.template()

    def push_primary_analyses_for_removal(self, analysis_request, analyses):
        """Stores the analyses to be removed after partitions creation
        """
        to_remove = self.analyses_to_remove.get(analysis_request, [])
        to_remove.extend(analyses)
        self.analyses_to_remove[analysis_request] = to_remove

    def remove_primary_analyses(self):
        """Remove analyses relocated to partitions
        """
        for ar, analyses in self.analyses_to_remove.items():
            analyses_ids = list(set(map(api.get_id, analyses)))
            ar.manage_delObjects(analyses_ids)
        self.analyses_to_remove = dict()

    def get_ar_data(self):
        """Returns a list of AR data
        """
        for obj in self.get_objects():
            info = self.get_base_info(obj)
            info.update({
                "analyses": self.get_analysis_data_for(obj),
                "sampletype": self.get_base_info(obj.getSampleType()),
                "number_of_partitions": self.get_number_of_partitions_for(obj),
                "template": self.get_template_data_for(obj),
            })
            yield info

    def get_sampletype_data(self):
        """Returns a list of SampleType data
        """
        for obj in self.get_sampletypes():
            info = self.get_base_info(obj)
            yield info

    def get_container_data(self):
        """Returns a list of Container data
        """
        for obj in self.get_containers():
            info = self.get_base_info(obj)
            yield info

    def get_preservation_data(self):
        """Returns a list of Preservation data
        """
        for obj in self.get_preservations():
            info = self.get_base_info(obj)
            yield info

    def get_objects(self):
        """Returns a list of objects coming from the "uids" request parameter
        """
        # Create a mapping of source ARs for copy
        uids = self.request.form.get("uids", "")
        if not uids:
            # check for the `items` parammeter
            uids = self.request.form.get("items", "")
        if isinstance(uids, basestring):
            uids = uids.split(",")
        unique_uids = OrderedDict().fromkeys(uids).keys()
        return filter(None, map(self.get_object_by_uid, unique_uids))

    def get_sampletypes(self):
        """Returns the available SampleTypes of the system
        """
        query = {
            "portal_type": "SampleType",
            "sort_on": "sortable_title",
            "sort_order": "ascending",
            "inactive_state": "active",
        }
        results = api.search(query, "bika_setup_catalog")
        return map(api.get_object, results)

    def get_containers(self):
        """Returns the available Containers of the system
        """
        query = dict(portal_type="Container",
                     sort_on="sortable_title",
                     sort_order="ascending",
                     inactive_state="active")
        results = api.search(query, "bika_setup_catalog")
        return map(api.get_object, results)

    def get_preservations(self):
        """Returns the available Preservations of the system
        """
        query = dict(portal_type="Preservation",
                     sort_on="sortable_title",
                     sort_order="ascending",
                     inactive_state="active")
        results = api.search(query, "bika_setup_catalog")
        return map(api.get_object, results)

    @returns_super_model
    def to_super_model(self, obj_or_objs):
        """Returns a SuperModel for a given object or a list of Supermodels if
        a list of objects was passed in
        """
        return obj_or_objs

    def get_analysis_data_for(self, ar):
        """Return the Analysis data for this AR
        """
        # Exclude analyses from children (partitions)
        analyses = ar.objectValues("Analysis")
        out = []
        for an in analyses:
            info = self.get_base_info(an)
            info.update({
                "service_uid": an.getServiceUID(),
            })
            out.append(info)
        return out

    def get_template_data_for(self, ar):
        """Return the Template data for this AR
        """
        info = None
        template = ar.getTemplate()
        ar_sampletype_uid = api.get_uid(ar.getSampleType())
        ar_container_uid = ""
        if ar.getContainer():
            ar_container_uid = api.get_uid(ar.getContainer())
        ar_preservation_uid = ""
        if ar.getPreservation():
            ar_preservation_uid = api.get_uid(ar.getPreservation())

        if template:
            info = self.get_base_info(template)

            analyses = template.getAnalyses()
            partition_analyses = map(
                lambda x: (x.get("partition"), x.get("service_uid")), analyses)

            analyses_by_partition = defaultdict(list)
            for partition, service_uid in partition_analyses:
                analyses_by_partition[partition].append(service_uid)

            sampletypes_by_partition = defaultdict(list)
            containers_by_partition = defaultdict(list)
            preservations_by_partition = defaultdict(list)
            for part in template.getPartitions():
                part_id = part.get("part_id")
                sampletype_uid = part.get('sampletype_uid', ar_sampletype_uid)
                sampletypes_by_partition[part_id] = sampletype_uid
                container_uid = part.get("container_uid", ar_container_uid)
                containers_by_partition[part_id] = container_uid
                preserv_uid = part.get("preservation_uid", ar_preservation_uid)
                preservations_by_partition[part_id] = preserv_uid

            partitions = map(lambda p: p.get("part_id"),
                             template.getPartitions())
            info.update({
                "analyses": analyses_by_partition,
                "partitions": partitions,
                "sample_types": sampletypes_by_partition,
                "containers": containers_by_partition,
                "preservations": preservations_by_partition,
            })
        else:
            info = {
                "analyses": {},
                "partitions": [],
                "sample_types": {},
                "containers": {},
                "preservations": {},
            }
        return info

    def get_number_of_partitions_for(self, ar):
        """Return the number of selected partitions
        """
        # fetch the number of partitions from the request
        uid = api.get_uid(ar)
        num = self.request.get("primary", {}).get(uid)

        if num is None:
            # get the number of partitions from the template
            template = ar.getTemplate()
            if template:
                num = len(template.getPartitions())
            else:
                num = DEFAULT_NUMBER_OF_PARTITIONS
        try:
            num = int(num)
        except (TypeError, ValueError):
            num = DEFAULT_NUMBER_OF_PARTITIONS
        return num

    def get_base_info(self, obj):
        """Extract the base info from the given object
        """
        obj = api.get_object(obj)
        review_state = api.get_workflow_status_of(obj)
        state_title = review_state.capitalize().replace("_", " ")
        return {
            "obj": obj,
            "id": api.get_id(obj),
            "uid": api.get_uid(obj),
            "title": api.get_title(obj),
            "path": api.get_path(obj),
            "url": api.get_url(obj),
            "review_state": review_state,
            "state_title": state_title,
        }

    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 get_object_by_uid(self, uid):
        """Get the object by UID
        """
        logger.debug("get_object_by_uid::UID={}".format(uid))
        obj = api.get_object_by_uid(uid, None)
        if obj is None:
            logger.warn("!! No object found for UID #{} !!")
        return obj

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


1			# -- coding: utf-8 --
2
3			from collections import OrderedDict
4			from collections import defaultdict
5
6			from Products.Five.browser import BrowserView
7			from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile
8			from bika.lims import api
9			from bika.lims import bikaMessageFactory as _
10			from bika.lims import logger
11			from bika.lims.decorators import returns_super_model
12			from bika.lims.utils.analysisrequest import create_partition
13
14			DEFAULT_NUMBER_OF_PARTITIONS = 0
15
16
17			class PartitionMagicView(BrowserView):
18			"""Manage Partitions of primary ARs
19			"""
20			template = ViewPageTemplateFile("templates/partition_magic.pt")
21
22			def __init__(self, context, request):
23			super(PartitionMagicView, self).__init__(context, request)
24			self.context = context
25			self.request = request
26			self.back_url = self.context.absolute_url()
27			self.analyses_to_remove = dict()
28
29
30			def __call__(self):
31			form = self.request.form
32
33			# Form submit toggle
34			form_submitted = form.get("submitted", False)
35
36			# Buttons
37			form_preview = form.get("button_preview", False)
38			form_create = form.get("button_create", False)
39			form_cancel = form.get("button_cancel", False)
40
41			objs = self.get_objects()
42
43			# No ARs selected
44			if not objs:
45			return self.redirect(message=_("No items selected"),
46			level="warning")
47
48			# Handle preview
49			if form_submitted and form_preview:
50			logger.info("* PREVIEW *")
51
52			# Handle create
53			if form_submitted and form_create:
54			logger.info("* CREATE PARTITIONS *")
55
56			partitions = []
57
58			# create the partitions
59			for partition in form.get("partitions", []):
60			primary_uid = partition.get("primary_uid")
61			sampletype_uid = partition.get("sampletype_uid")
62			container_uid = partition.get("container_uid")
63			preservation_uid = partition.get("preservation_uid")
64			if not primary_uid:
65			continue
66
67			# The creation of partitions w/o analyses is allowed. Maybe the
68			# user wants to add the analyses later manually or wants to keep
69			# this partition stored in a freezer for some time
70			# Note we set "remove_primary_analyses" to False cause we want
71			# user to be able to add same analyses to different partitions.
72			analyses_uids = partition.get("analyses", [])
73			partition = create_partition(
74			request=self.request,
75			analysis_request=primary_uid,
76			sample_type=sampletype_uid,
77			container=container_uid,
78			preservation=preservation_uid,
79			analyses=analyses_uids,
80			remove_primary_analyses=False,
81			)
82			partitions.append(partition)
83
84			logger.info("Successfully created partition: {}".format(
85			api.get_path(partition)))
86
87			if not partitions:
88			# If no partitions were created, show a warning message
89			return self.redirect(message=_("No partitions were created"))
90
91			# Remove analyses from primary Analysis Requests
92			self.remove_primary_analyses()
93
94			message = _("Created {} partitions: {}".format(
95			len(partitions), ", ".join(map(api.get_title, partitions))))
96			return self.redirect(message=message)
97
98			# Handle cancel
99			if form_submitted and form_cancel:
100			logger.info("* CANCEL *")
101			return self.redirect(message=_("Partitioning canceled"))
102
103			return self.template()
104
105			def push_primary_analyses_for_removal(self, analysis_request, analyses):
106			"""Stores the analyses to be removed after partitions creation
107			"""
108			to_remove = self.analyses_to_remove.get(analysis_request, [])
109			to_remove.extend(analyses)
110			self.analyses_to_remove[analysis_request] = to_remove
111
112			def remove_primary_analyses(self):
113			"""Remove analyses relocated to partitions
114			"""
115			for ar, analyses in self.analyses_to_remove.items():
116			analyses_ids = list(set(map(api.get_id, analyses)))
117			ar.manage_delObjects(analyses_ids)
118			self.analyses_to_remove = dict()
119
120			def get_ar_data(self):
121			"""Returns a list of AR data
122			"""
123			for obj in self.get_objects():
124			info = self.get_base_info(obj)
125			info.update({
126			"analyses": self.get_analysis_data_for(obj),
127			"sampletype": self.get_base_info(obj.getSampleType()),
128			"number_of_partitions": self.get_number_of_partitions_for(obj),
129			"template": self.get_template_data_for(obj),
130			})
131			yield info
132
133			def get_sampletype_data(self):
134			"""Returns a list of SampleType data
135			"""
136			for obj in self.get_sampletypes():
137			info = self.get_base_info(obj)
138			yield info
139
140			def get_container_data(self):
141			"""Returns a list of Container data
142			"""
143			for obj in self.get_containers():
144			info = self.get_base_info(obj)
145			yield info
146
147			def get_preservation_data(self):
148			"""Returns a list of Preservation data
149			"""
150			for obj in self.get_preservations():
151			info = self.get_base_info(obj)
152			yield info
153
154			def get_objects(self):
155			"""Returns a list of objects coming from the "uids" request parameter
156			"""
157			# Create a mapping of source ARs for copy
158			uids = self.request.form.get("uids", "")
159			if not uids:
160			# check for the `items` parammeter
161			uids = self.request.form.get("items", "")
162			if isinstance(uids, basestring):
163			uids = uids.split(",")
164			unique_uids = OrderedDict().fromkeys(uids).keys()
165			return filter(None, map(self.get_object_by_uid, unique_uids))
166
167			def get_sampletypes(self):
168			"""Returns the available SampleTypes of the system
169			"""
170			query = {
171			"portal_type": "SampleType",
172			"sort_on": "sortable_title",
173			"sort_order": "ascending",
174			"inactive_state": "active",
175			}
176			results = api.search(query, "bika_setup_catalog")
177			return map(api.get_object, results)
178
179			def get_containers(self):
180			"""Returns the available Containers of the system
181			"""
182			query = dict(portal_type="Container",
183			sort_on="sortable_title",
184			sort_order="ascending",
185			inactive_state="active")
186			results = api.search(query, "bika_setup_catalog")
187			return map(api.get_object, results)
188
189			def get_preservations(self):
190			"""Returns the available Preservations of the system
191			"""
192			query = dict(portal_type="Preservation",
193			sort_on="sortable_title",
194			sort_order="ascending",
195			inactive_state="active")
196			results = api.search(query, "bika_setup_catalog")
197			return map(api.get_object, results)
198
199			@returns_super_model
200			def to_super_model(self, obj_or_objs):
201			"""Returns a SuperModel for a given object or a list of Supermodels if
202			a list of objects was passed in
203			"""
204			return obj_or_objs
205
206			def get_analysis_data_for(self, ar):
207			"""Return the Analysis data for this AR
208			"""
209			# Exclude analyses from children (partitions)
210			analyses = ar.objectValues("Analysis")
211			out = []
212			for an in analyses:
213			info = self.get_base_info(an)
214			info.update({
215			"service_uid": an.getServiceUID(),
216			})
217			out.append(info)
218			return out
219
220			def get_template_data_for(self, ar):
221			"""Return the Template data for this AR
222			"""
223			info = None
224			template = ar.getTemplate()
225			ar_sampletype_uid = api.get_uid(ar.getSampleType())
226			ar_container_uid = ""
227			if ar.getContainer():
228			ar_container_uid = api.get_uid(ar.getContainer())
229			ar_preservation_uid = ""
230			if ar.getPreservation():
231			ar_preservation_uid = api.get_uid(ar.getPreservation())
232
233			if template:
234			info = self.get_base_info(template)
235
236			analyses = template.getAnalyses()
237			partition_analyses = map(
238			lambda x: (x.get("partition"), x.get("service_uid")), analyses)
239
240			analyses_by_partition = defaultdict(list)
241			for partition, service_uid in partition_analyses:
242			analyses_by_partition[partition].append(service_uid)
243
244			sampletypes_by_partition = defaultdict(list)
245			containers_by_partition = defaultdict(list)
246			preservations_by_partition = defaultdict(list)
247			for part in template.getPartitions():
248			part_id = part.get("part_id")
249			sampletype_uid = part.get('sampletype_uid', ar_sampletype_uid)
250			sampletypes_by_partition[part_id] = sampletype_uid
251			container_uid = part.get("container_uid", ar_container_uid)
252			containers_by_partition[part_id] = container_uid
253			preserv_uid = part.get("preservation_uid", ar_preservation_uid)
254			preservations_by_partition[part_id] = preserv_uid
255
256			partitions = map(lambda p: p.get("part_id"),
257			template.getPartitions())
258			info.update({
259			"analyses": analyses_by_partition,
260			"partitions": partitions,
261			"sample_types": sampletypes_by_partition,
262			"containers": containers_by_partition,
263			"preservations": preservations_by_partition,
264			})
265			else:
266			info = {
267			"analyses": {},
268			"partitions": [],
269			"sample_types": {},
270			"containers": {},
271			"preservations": {},
272			}
273			return info
274
275			def get_number_of_partitions_for(self, ar):
276			"""Return the number of selected partitions
277			"""
278			# fetch the number of partitions from the request
279			uid = api.get_uid(ar)
280			num = self.request.get("primary", {}).get(uid)
281
282			if num is None:
283			# get the number of partitions from the template
284			template = ar.getTemplate()
285			if template:
286			num = len(template.getPartitions())
287			else:
288			num = DEFAULT_NUMBER_OF_PARTITIONS
289			try:
290			num = int(num)
291			except (TypeError, ValueError):
292			num = DEFAULT_NUMBER_OF_PARTITIONS
293			return num
294
295			def get_base_info(self, obj):
296			"""Extract the base info from the given object
297			"""
298			obj = api.get_object(obj)
299			review_state = api.get_workflow_status_of(obj)
300			state_title = review_state.capitalize().replace("_", " ")
301			return {
302			"obj": obj,
303			"id": api.get_id(obj),
304			"uid": api.get_uid(obj),
305			"title": api.get_title(obj),
306			"path": api.get_path(obj),
307			"url": api.get_url(obj),
308			"review_state": review_state,
309			"state_title": state_title,
310			}
311
312			def redirect(self, redirect_url=None, message=None, level="info"):
313			"""Redirect with a message
314			"""
315			if redirect_url is None:
316			redirect_url = self.back_url
317			if message is not None:
318			self.add_status_message(message, level)
319			return self.request.response.redirect(redirect_url)
320
321			def get_object_by_uid(self, uid):
322			"""Get the object by UID
323			"""
324			logger.debug("get_object_by_uid::UID={}".format(uid))
325			obj = api.get_object_by_uid(uid, None)
326			if obj is None:
327			logger.warn("!! No object found for UID #{} !!")
328			return obj
329
330			def add_status_message(self, message, level="info"):
331			"""Set a portal status message
332			"""
333			return self.context.plone_utils.addPortalMessage(message, level)
334

senaite / senaite.core

Push — master ( ac1405...5f2185 )

PartitionMagicView.get_containers() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like