bika.lims.browser.partition_magic.PartitionMagicView.to_super_model() - Code Metrics - Inspection of "Allow the creation of empty partitions (without an..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( b7ddba...ac1405 )

by Ramon

created 2019-02-20 07:18 UTC

PartitionMagicView.to_super_model() A

↳ Parent: bika.lims.browser.partition_magic

Complexity

Conditions

Size

Total Lines	6
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	3
dl	0
loc	6
rs	10
c	0
b	0
f	0
cc	1
nop	2

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

from collections import OrderedDict
from collections import defaultdict

from bika.lims import bikaMessageFactory as _
from bika.lims import logger
from bika.lims.decorators import returns_super_model
from bika.lims import workflow as wf
from bika.lims import api
from bika.lims.interfaces import IProxyField
from bika.lims.utils.analysisrequest import create_analysisrequest as crar
from Products.Five.browser import BrowserView
from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile

DEFAULT_NUMBER_OF_PARTITIONS = 0

PARTITION_SKIP_FIELDS = [
    "Analyses",
    "Attachment",
    "Client",
    "Container",
    "Preservation",
    "Profile",
    "Profiles",
    "RejectionReasons",
    "Remarks",
    "ResultsInterpretation",
    "ResultsInterpretationDepts",
    "Sample",
    "SampleType",
    "Template",
    "creation_date",
    "id",
    "modification_date",
    "ParentAnalysisRequest",
]


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
                analyses_uids = partition.get("analyses", [])

                partition = self.create_partition(
                    primary_uid=primary_uid,
                    sampletype_uid=sampletype_uid,
                    container_uid=container_uid,
                    preservation_uid=preservation_uid,
                    analyses_uids=analyses_uids)
                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 create_partition(self, primary_uid, sampletype_uid, container_uid,
                         preservation_uid, analyses_uids):
        """Create a new partition (AR)
        """
        logger.info("*** CREATE PARTITION ***")

        ar = self.get_object_by_uid(primary_uid)
        record = {
            "InternalUse": True,
            "ParentAnalysisRequest": primary_uid,
            "SampleType": sampletype_uid,
            "Container": container_uid,
            "Preservation": preservation_uid,
        }

        for fieldname, field in api.get_fields(ar).items():
            # if self.is_proxy_field(field):
            #     logger.info("Skipping proxy field {}".format(fieldname))
            #     continue
            if self.is_computed_field(field):
                logger.info("Skipping computed field {}".format(fieldname))
                continue
            if fieldname in PARTITION_SKIP_FIELDS:
                logger.info("Skipping field {}".format(fieldname))
                continue
            fieldvalue = field.get(ar)
            record[fieldname] = fieldvalue
            logger.info("Update record '{}': {}".format(
                fieldname, repr(fieldvalue)))

        client = ar.getClient()
        analyses = map(self.get_object_by_uid, analyses_uids)
        services = map(lambda an: an.getAnalysisService(), analyses)

        partition = crar(
            client,
            self.request,
            record,
            analyses=services,
            specifications=self.get_specifications_for(ar)
        )

        # Remove selected analyses from the parent Analysis Request
        self.push_primary_analyses_for_removal(ar, analyses)

        # Reindex Parent Analysis Request
        ar.reindexObject(idxs=["isRootAncestor"])

        # Manually set the Date Received to match with its parent. This is
        # necessary because crar calls to processForm, so DateReceived is not
        # set because the partition has not been received yet
        partition.setDateReceived(ar.getDateReceived())
        partition.reindexObject(idxs="getDateReceived")

        # Force partition to same status as the primary
        status = api.get_workflow_status_of(ar)
        wf.changeWorkflowState(partition, "bika_ar_workflow", status)

        # And initialize the analyses the partition contains. This is required
        # here because the transition "initialize" of analyses rely on a guard,
        # so the initialization can only be performed when the sample has been
        # received (DateReceived is set)
        wf.ActionHandlerPool.get_instance().queue_pool()
        for analysis in partition.getAnalyses(full_objects=True):
            wf.doActionFor(analysis, "initialize")
        wf.ActionHandlerPool.get_instance().resume()
        return partition

    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_specifications_for(self, ar):
        """Returns a mapping of service uid -> specification
        """
        spec = ar.getSpecification()
        if not spec:
            return []
        return spec.getResultsRange()

    def is_proxy_field(self, field):
        """Checks if the field is a proxy field
        """
        return IProxyField.providedBy(field)

    def is_computed_field(self, field):
        """Checks if the field is a coumputed field
        """
        return field.type == "computed"

    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 bika.lims import bikaMessageFactory as _
7			from bika.lims import logger
8			from bika.lims.decorators import returns_super_model
9			from bika.lims import workflow as wf
10			from bika.lims import api
11			from bika.lims.interfaces import IProxyField
12			from bika.lims.utils.analysisrequest import create_analysisrequest as crar
13			from Products.Five.browser import BrowserView
14			from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile
15
16			DEFAULT_NUMBER_OF_PARTITIONS = 0
17
18			PARTITION_SKIP_FIELDS = [
19			"Analyses",
20			"Attachment",
21			"Client",
22			"Container",
23			"Preservation",
24			"Profile",
25			"Profiles",
26			"RejectionReasons",
27			"Remarks",
28			"ResultsInterpretation",
29			"ResultsInterpretationDepts",
30			"Sample",
31			"SampleType",
32			"Template",
33			"creation_date",
34			"id",
35			"modification_date",
36			"ParentAnalysisRequest",
37			]
38
39
40			class PartitionMagicView(BrowserView):
41			"""Manage Partitions of primary ARs
42			"""
43			template = ViewPageTemplateFile("templates/partition_magic.pt")
44
45			def __init__(self, context, request):
46			super(PartitionMagicView, self).__init__(context, request)
47			self.context = context
48			self.request = request
49			self.back_url = self.context.absolute_url()
50			self.analyses_to_remove = dict()
51
52
53			def __call__(self):
54			form = self.request.form
55
56			# Form submit toggle
57			form_submitted = form.get("submitted", False)
58
59			# Buttons
60			form_preview = form.get("button_preview", False)
61			form_create = form.get("button_create", False)
62			form_cancel = form.get("button_cancel", False)
63
64			objs = self.get_objects()
65
66			# No ARs selected
67			if not objs:
68			return self.redirect(message=_("No items selected"),
69			level="warning")
70
71			# Handle preview
72			if form_submitted and form_preview:
73			logger.info("* PREVIEW *")
74
75			# Handle create
76			if form_submitted and form_create:
77			logger.info("* CREATE PARTITIONS *")
78
79			partitions = []
80
81			# create the partitions
82			for partition in form.get("partitions", []):
83			primary_uid = partition.get("primary_uid")
84			sampletype_uid = partition.get("sampletype_uid")
85			container_uid = partition.get("container_uid")
86			preservation_uid = partition.get("preservation_uid")
87			if not primary_uid:
88			continue
89
90			# The creation of partitions w/o analyses is allowed. Maybe the
91			# user wants to add the analyses later manually or wants to keep
92			# this partition stored in a freezer for some time
93			analyses_uids = partition.get("analyses", [])
94
95			partition = self.create_partition(
96			primary_uid=primary_uid,
97			sampletype_uid=sampletype_uid,
98			container_uid=container_uid,
99			preservation_uid=preservation_uid,
100			analyses_uids=analyses_uids)
101			partitions.append(partition)
102
103			logger.info("Successfully created partition: {}".format(
104			api.get_path(partition)))
105
106			if not partitions:
107			# If no partitions were created, show a warning message
108			return self.redirect(message=_("No partitions were created"))
109
110			# Remove analyses from primary Analysis Requests
111			self.remove_primary_analyses()
112
113			message = _("Created {} partitions: {}".format(
114			len(partitions), ", ".join(map(api.get_title, partitions))))
115			return self.redirect(message=message)
116
117			# Handle cancel
118			if form_submitted and form_cancel:
119			logger.info("* CANCEL *")
120			return self.redirect(message=_("Partitioning canceled"))
121
122			return self.template()
123
124			def create_partition(self, primary_uid, sampletype_uid, container_uid,
125			preservation_uid, analyses_uids):
126			"""Create a new partition (AR)
127			"""
128			logger.info("* CREATE PARTITION *")
129
130			ar = self.get_object_by_uid(primary_uid)
131			record = {
132			"InternalUse": True,
133			"ParentAnalysisRequest": primary_uid,
134			"SampleType": sampletype_uid,
135			"Container": container_uid,
136			"Preservation": preservation_uid,
137			}
138
139			for fieldname, field in api.get_fields(ar).items():
140			# if self.is_proxy_field(field):
141			# logger.info("Skipping proxy field {}".format(fieldname))
142			# continue
143			if self.is_computed_field(field):
144			logger.info("Skipping computed field {}".format(fieldname))
145			continue
146			if fieldname in PARTITION_SKIP_FIELDS:
147			logger.info("Skipping field {}".format(fieldname))
148			continue
149			fieldvalue = field.get(ar)
150			record[fieldname] = fieldvalue
151			logger.info("Update record '{}': {}".format(
152			fieldname, repr(fieldvalue)))
153
154			client = ar.getClient()
155			analyses = map(self.get_object_by_uid, analyses_uids)
156			services = map(lambda an: an.getAnalysisService(), analyses)
157
158			partition = crar(
159			client,
160			self.request,
161			record,
162			analyses=services,
163			specifications=self.get_specifications_for(ar)
164			)
165
166			# Remove selected analyses from the parent Analysis Request
167			self.push_primary_analyses_for_removal(ar, analyses)
168
169			# Reindex Parent Analysis Request
170			ar.reindexObject(idxs=["isRootAncestor"])
171
172			# Manually set the Date Received to match with its parent. This is
173			# necessary because crar calls to processForm, so DateReceived is not
174			# set because the partition has not been received yet
175			partition.setDateReceived(ar.getDateReceived())
176			partition.reindexObject(idxs="getDateReceived")
177
178			# Force partition to same status as the primary
179			status = api.get_workflow_status_of(ar)
180			wf.changeWorkflowState(partition, "bika_ar_workflow", status)
181
182			# And initialize the analyses the partition contains. This is required
183			# here because the transition "initialize" of analyses rely on a guard,
184			# so the initialization can only be performed when the sample has been
185			# received (DateReceived is set)
186			wf.ActionHandlerPool.get_instance().queue_pool()
187			for analysis in partition.getAnalyses(full_objects=True):
188			wf.doActionFor(analysis, "initialize")
189			wf.ActionHandlerPool.get_instance().resume()
190			return partition
191
192			def push_primary_analyses_for_removal(self, analysis_request, analyses):
193			"""Stores the analyses to be removed after partitions creation
194			"""
195			to_remove = self.analyses_to_remove.get(analysis_request, [])
196			to_remove.extend(analyses)
197			self.analyses_to_remove[analysis_request] = to_remove
198
199			def remove_primary_analyses(self):
200			"""Remove analyses relocated to partitions
201			"""
202			for ar, analyses in self.analyses_to_remove.items():
203			analyses_ids = list(set(map(api.get_id, analyses)))
204			ar.manage_delObjects(analyses_ids)
205			self.analyses_to_remove = dict()
206
207			def get_specifications_for(self, ar):
208			"""Returns a mapping of service uid -> specification
209			"""
210			spec = ar.getSpecification()
211			if not spec:
212			return []
213			return spec.getResultsRange()
214
215			def is_proxy_field(self, field):
216			"""Checks if the field is a proxy field
217			"""
218			return IProxyField.providedBy(field)
219
220			def is_computed_field(self, field):
221			"""Checks if the field is a coumputed field
222			"""
223			return field.type == "computed"
224
225			def get_ar_data(self):
226			"""Returns a list of AR data
227			"""
228			for obj in self.get_objects():
229			info = self.get_base_info(obj)
230			info.update({
231			"analyses": self.get_analysis_data_for(obj),
232			"sampletype": self.get_base_info(obj.getSampleType()),
233			"number_of_partitions": self.get_number_of_partitions_for(obj),
234			"template": self.get_template_data_for(obj),
235			})
236			yield info
237
238			def get_sampletype_data(self):
239			"""Returns a list of SampleType data
240			"""
241			for obj in self.get_sampletypes():
242			info = self.get_base_info(obj)
243			yield info
244
245			def get_container_data(self):
246			"""Returns a list of Container data
247			"""
248			for obj in self.get_containers():
249			info = self.get_base_info(obj)
250			yield info
251
252			def get_preservation_data(self):
253			"""Returns a list of Preservation data
254			"""
255			for obj in self.get_preservations():
256			info = self.get_base_info(obj)
257			yield info
258
259			def get_objects(self):
260			"""Returns a list of objects coming from the "uids" request parameter
261			"""
262			# Create a mapping of source ARs for copy
263			uids = self.request.form.get("uids", "")
264			if not uids:
265			# check for the `items` parammeter
266			uids = self.request.form.get("items", "")
267			if isinstance(uids, basestring):
268			uids = uids.split(",")
269			unique_uids = OrderedDict().fromkeys(uids).keys()
270			return filter(None, map(self.get_object_by_uid, unique_uids))
271
272			def get_sampletypes(self):
273			"""Returns the available SampleTypes of the system
274			"""
275			query = {
276			"portal_type": "SampleType",
277			"sort_on": "sortable_title",
278			"sort_order": "ascending",
279			"inactive_state": "active",
280			}
281			results = api.search(query, "bika_setup_catalog")
282			return map(api.get_object, results)
283
284			def get_containers(self):
285			"""Returns the available Containers of the system
286			"""
287			query = dict(portal_type="Container",
288			sort_on="sortable_title",
289			sort_order="ascending",
290			inactive_state="active")
291			results = api.search(query, "bika_setup_catalog")
292			return map(api.get_object, results)
293
294			def get_preservations(self):
295			"""Returns the available Preservations of the system
296			"""
297			query = dict(portal_type="Preservation",
298			sort_on="sortable_title",
299			sort_order="ascending",
300			inactive_state="active")
301			results = api.search(query, "bika_setup_catalog")
302			return map(api.get_object, results)
303
304			@returns_super_model
305			def to_super_model(self, obj_or_objs):
306			"""Returns a SuperModel for a given object or a list of Supermodels if
307			a list of objects was passed in
308			"""
309			return obj_or_objs
310
311			def get_analysis_data_for(self, ar):
312			"""Return the Analysis data for this AR
313			"""
314			# Exclude analyses from children (partitions)
315			analyses = ar.objectValues("Analysis")
316			out = []
317			for an in analyses:
318			info = self.get_base_info(an)
319			info.update({
320			"service_uid": an.getServiceUID(),
321			})
322			out.append(info)
323			return out
324
325			def get_template_data_for(self, ar):
326			"""Return the Template data for this AR
327			"""
328			info = None
329			template = ar.getTemplate()
330			ar_sampletype_uid = api.get_uid(ar.getSampleType())
331			ar_container_uid = ""
332			if ar.getContainer():
333			ar_container_uid = api.get_uid(ar.getContainer())
334			ar_preservation_uid = ""
335			if ar.getPreservation():
336			ar_preservation_uid = api.get_uid(ar.getPreservation())
337
338			if template:
339			info = self.get_base_info(template)
340
341			analyses = template.getAnalyses()
342			partition_analyses = map(
343			lambda x: (x.get("partition"), x.get("service_uid")), analyses)
344
345			analyses_by_partition = defaultdict(list)
346			for partition, service_uid in partition_analyses:
347			analyses_by_partition[partition].append(service_uid)
348
349			sampletypes_by_partition = defaultdict(list)
350			containers_by_partition = defaultdict(list)
351			preservations_by_partition = defaultdict(list)
352			for part in template.getPartitions():
353			part_id = part.get("part_id")
354			sampletype_uid = part.get('sampletype_uid', ar_sampletype_uid)
355			sampletypes_by_partition[part_id] = sampletype_uid
356			container_uid = part.get("container_uid", ar_container_uid)
357			containers_by_partition[part_id] = container_uid
358			preserv_uid = part.get("preservation_uid", ar_preservation_uid)
359			preservations_by_partition[part_id] = preserv_uid
360
361			partitions = map(lambda p: p.get("part_id"),
362			template.getPartitions())
363			info.update({
364			"analyses": analyses_by_partition,
365			"partitions": partitions,
366			"sample_types": sampletypes_by_partition,
367			"containers": containers_by_partition,
368			"preservations": preservations_by_partition,
369			})
370			else:
371			info = {
372			"analyses": {},
373			"partitions": [],
374			"sample_types": {},
375			"containers": {},
376			"preservations": {},
377			}
378			return info
379
380			def get_number_of_partitions_for(self, ar):
381			"""Return the number of selected partitions
382			"""
383			# fetch the number of partitions from the request
384			uid = api.get_uid(ar)
385			num = self.request.get("primary", {}).get(uid)
386
387			if num is None:
388			# get the number of partitions from the template
389			template = ar.getTemplate()
390			if template:
391			num = len(template.getPartitions())
392			else:
393			num = DEFAULT_NUMBER_OF_PARTITIONS
394			try:
395			num = int(num)
396			except (TypeError, ValueError):
397			num = DEFAULT_NUMBER_OF_PARTITIONS
398			return num
399
400			def get_base_info(self, obj):
401			"""Extract the base info from the given object
402			"""
403			obj = api.get_object(obj)
404			review_state = api.get_workflow_status_of(obj)
405			state_title = review_state.capitalize().replace("_", " ")
406			return {
407			"obj": obj,
408			"id": api.get_id(obj),
409			"uid": api.get_uid(obj),
410			"title": api.get_title(obj),
411			"path": api.get_path(obj),
412			"url": api.get_url(obj),
413			"review_state": review_state,
414			"state_title": state_title,
415			}
416
417			def redirect(self, redirect_url=None, message=None, level="info"):
418			"""Redirect with a message
419			"""
420			if redirect_url is None:
421			redirect_url = self.back_url
422			if message is not None:
423			self.add_status_message(message, level)
424			return self.request.response.redirect(redirect_url)
425
426			def get_object_by_uid(self, uid):
427			"""Get the object by UID
428			"""
429			logger.debug("get_object_by_uid::UID={}".format(uid))
430			obj = api.get_object_by_uid(uid, None)
431			if obj is None:
432			logger.warn("!! No object found for UID #{} !!")
433			return obj
434
435			def add_status_message(self, message, level="info"):
436			"""Set a portal status message
437			"""
438			return self.context.plone_utils.addPortalMessage(message, level)
439

senaite / senaite.core

Push — master ( b7ddba...ac1405 )

PartitionMagicView.to_super_model() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like