bika.lims.browser.fields.aranalysesfield.ARAnalysesField.get_from_descendant() - Code Metrics - Inspection of "Ensure analysis' result range keeps all range subf..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 2d15ca...15a906 )

by Ramon

created 2020-02-02 20:48 UTC

ARAnalysesField.get_from_descendant() A

↳ Parent: bika.lims.browser.fields.aranalysesfield

Complexity

Conditions

Size

Total Lines	15
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	9
dl	0
loc	15
rs	9.95
c	0
b	0
f	0
cc	3
nop	3

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE.
#
# SENAITE.CORE is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, version 2.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright 2018-2019 by it's authors.
# Some rights reserved, see README and LICENSE.

import itertools

from AccessControl import ClassSecurityInfo
from AccessControl import Unauthorized
from Products.Archetypes.Registry import registerField
from Products.Archetypes.public import Field
from Products.Archetypes.public import ObjectField
from zope.interface import alsoProvides
from zope.interface import implements
from zope.interface import noLongerProvides

from bika.lims import api
from bika.lims import logger
from bika.lims.api.security import check_permission
from bika.lims.catalog import CATALOG_ANALYSIS_LISTING
from bika.lims.catalog import SETUP_CATALOG
from bika.lims.interfaces import IARAnalysesField
from bika.lims.interfaces import IAnalysis
from bika.lims.interfaces import IAnalysisService
from bika.lims.interfaces import IInternalUse
from bika.lims.interfaces import ISubmitted
from bika.lims.permissions import AddAnalysis
from bika.lims.utils.analysis import create_analysis

"""Field to manage Analyses on ARs

Please see the assigned doctest at tests/doctests/ARAnalysesField.rst

Run this test from the buildout directory:

    bin/test test_textual_doctests -t ARAnalysesField
"""


class ARAnalysesField(ObjectField):
    """A field that stores Analyses instances
    """
    implements(IARAnalysesField)

    security = ClassSecurityInfo()
    _properties = Field._properties.copy()
    _properties.update({
        "type": "analyses",
        "default": None,
    })

    security.declarePrivate('get')

    def get(self, instance, **kwargs):
        """Returns a list of Analyses assigned to this AR

        Return a list of catalog brains unless `full_objects=True` is passed.
        Other keyword arguments are passed to bika_analysis_catalog

        :param instance: Analysis Request object
        :param kwargs: Keyword arguments to inject in the search query
        :returns: A list of Analysis Objects/Catalog Brains
        """
        # Do we need to return objects or brains
        full_objects = kwargs.get("full_objects", False)

        # Bail out parameters from kwargs that don't match with indexes
        catalog = api.get_tool(CATALOG_ANALYSIS_LISTING)
        indexes = catalog.indexes()
        query = dict([(k, v) for k, v in kwargs.items() if k in indexes])

        # Do the search against the catalog
        query["portal_type"] = "Analysis"
        query["getAncestorsUIDs"] = api.get_uid(instance)
        brains = catalog(query)
        if full_objects:
            return map(api.get_object, brains)
        return brains

    security.declarePrivate('set')

    def set(self, instance, items, prices=None, specs=None, hidden=None, **kw):
        """Set/Assign Analyses to this AR

        :param items: List of Analysis objects/brains, AnalysisService
                      objects/brains and/or Analysis Service uids
        :type items: list
        :param prices: Mapping of AnalysisService UID -> price
        :type prices: dict
        :param specs: List of AnalysisService UID -> Result Range mappings
        :type specs: list
        :param hidden: List of AnalysisService UID -> Hidden mappings
        :type hidden: list
        :returns: list of new assigned Analyses
        """
        if items is None:
            items = []

        # Bail out if the items is not a list type
        if not isinstance(items, (list, tuple)):
            raise TypeError(
                "Items parameter must be a tuple or list, got '{}'".format(
                    type(items)))

        # Bail out if the AR is inactive
        if not api.is_active(instance):
            raise Unauthorized("Inactive ARs can not be modified"
                               .format(AddAnalysis))

        # Bail out if the user has not the right permission
        if not check_permission(AddAnalysis, instance):
            raise Unauthorized("You do not have the '{}' permission"
                               .format(AddAnalysis))

        # Convert the items to a valid list of AnalysisServices
        services = filter(None, map(self._to_service, items))

        # Calculate dependencies
        dependencies = map(lambda s: s.getServiceDependencies(), services)
        dependencies = list(itertools.chain.from_iterable(dependencies))

        # Merge dependencies and services
        services = set(services + dependencies)

        # Modify existing AR specs with new form values of selected analyses
        specs = self.resolve_specs(instance, specs)

        # Add analyses
        params = dict(prices=prices, hidden=hidden, specs=specs)
        map(lambda serv: self.add_analysis(instance, serv, **params), services)

        # Get all analyses (those from descendants included)
        analyses = instance.objectValues("Analysis")
        analyses.extend(self.get_analyses_from_descendants(instance))

        # Bail out those not in services list or submitted
        uids = map(api.get_uid, services)
        to_remove = filter(lambda an: an.getServiceUID() not in uids, analyses)
        to_remove = filter(lambda an: not ISubmitted.providedBy(an), to_remove)

        # Remove analyses
        map(self.remove_analysis, to_remove)

    def resolve_specs(self, instance, results_ranges):
        """Returns a dictionary where the key is the service_uid and the value
        is its results range. The dictionary is made by extending the
        results_ranges passed-in with the Sample's ResultsRanges (a copy of the
        specifications initially set)
        """
        rrs = results_ranges or []

        # Sample's Results ranges
        sample_rrs = instance.getResultsRange()

        # Ensure all subfields from specification are kept and missing values
        # for subfields are filled in accordance with the specs
        rrs = map(lambda rr: self.resolve_range(rr, sample_rrs), rrs)

        # Append those from sample that are missing in the ranges passed-in
        service_uids = map(lambda rr: rr["uid"], rrs)
        rrs.extend(filter(lambda rr: rr["uid"] not in service_uids, sample_rrs))

        # Create a dict for easy access to results ranges
        return dict(map(lambda rr: (rr["uid"], rr), rrs))

    def resolve_range(self, result_range, sample_result_ranges):
        """Resolves the range by adding the uid if not present and filling the
        missing subfield values with those that come from the Sample
        specification if they are not present in the result_range passed-in
        """
        # Resolve result_range to make sure it contain uid subfield
        rrs = self.resolve_uid(result_range)
        uid = rrs.get("uid")

        for sample_rr in sample_result_ranges:
            if uid and sample_rr.get("uid") == uid:
                # Keep same fields from sample
                rr = sample_rr.copy()
                rr.update(rrs)
                return rr

        # Return the original with no changes
        return rrs

    def resolve_uid(self, result_range):

        """Resolves the uid key for the result_range passed in if it does not
        exist when contains a keyword
        """
        value = result_range.copy()
        uid = value.get("uid")
        if api.is_uid(uid) and uid != "0":
            return value

        # uid key does not exist or is not valid, try to infere from keyword
        keyword = value.get("keyword")
        if keyword:
            query = dict(portal_type="AnalysisService", getKeyword=keyword)
            brains = api.search(query, SETUP_CATALOG)
            if len(brains) == 1:
                uid = api.get_uid(brains[0])
        value["uid"] = uid
        return value

    def add_analysis(self, instance, service, **kwargs):
        service_uid = api.get_uid(service)

        # Ensure we have suitable parameters
        specs = kwargs.get("specs") or {}

        # Get the hidden status for the service
        hidden = kwargs.get("hidden") or []
        hidden = filter(lambda d: d.get("uid") == service_uid, hidden)
        hidden = hidden and hidden[0].get("hidden") or service.getHidden()

        # Get the price for the service
        prices = kwargs.get("prices") or {}
        price = prices.get(service_uid) or service.getPrice()

        # Gets the analysis or creates the analysis for this service
        # Note this returns a list, because is possible to have multiple
        # partitions with same analysis
        analyses = self.resolve_analyses(instance, service)
        if not analyses:
            # Create the analysis
            keyword = service.getKeyword()
            logger.info("Creating new analysis '{}'".format(keyword))
            analysis = create_analysis(instance, service)
            analyses.append(analysis)

        skip = ["cancelled", "retracted", "rejected"]
        for analysis in analyses:
            # Skip analyses to better not modify
            if api.get_review_status(analysis) in skip:
                continue

            # Set the hidden status
            analysis.setHidden(hidden)

            # Set the price of the Analysis
            analysis.setPrice(price)

            # Set the internal use status
            parent_sample = analysis.getRequest()
            analysis.setInternalUse(parent_sample.getInternalUse())

            # Set the result range to the analysis
            analysis_rr = specs.get(service_uid) or analysis.getResultsRange()
            analysis.setResultsRange(analysis_rr)
            analysis.reindexObject()

    def remove_analysis(self, analysis):
        """Removes a given analysis from the instance
        """
        # Remember assigned attachments
        # https://github.com/senaite/senaite.core/issues/1025
        attachments = analysis.getAttachment()
        analysis.setAttachment([])

        # If assigned to a worksheet, unassign it before deletion
        worksheet = analysis.getWorksheet()
        if worksheet:
            worksheet.removeAnalysis(analysis)

        # Remove the analysis
        # Note the analysis might belong to a partition
        analysis.aq_parent.manage_delObjects(ids=[api.get_id(analysis)])

        # Remove orphaned attachments
        for attachment in attachments:
            if not attachment.getLinkedAnalyses():
                # only delete attachments which are no further linked
                logger.info(
                    "Deleting attachment: {}".format(attachment.getId()))
                attachment_id = api.get_id(attachment)
                api.get_parent(attachment).manage_delObjects(attachment_id)

    def resolve_analyses(self, instance, service):
        """Resolves analyses for the service and instance
        It returns a list, cause for a given sample, multiple analyses for same
        service can exist due to the possibility of having multiple partitions
        """
        analyses = []

        # Does the analysis exists in this instance already?
        instance_analyses = self.get_from_instance(instance, service)
        if instance_analyses:
            analyses.extend(instance_analyses)

        # Does the analysis exists in an ancestor?
        from_ancestor = self.get_from_ancestor(instance, service)
        for ancestor_analysis in from_ancestor:
            # Move the analysis into this instance. The ancestor's
            # analysis will be masked otherwise
            analysis_id = api.get_id(ancestor_analysis)
            logger.info("Analysis {} is from an ancestor".format(analysis_id))
            cp = ancestor_analysis.aq_parent.manage_cutObjects(analysis_id)
            instance.manage_pasteObjects(cp)
            analyses.append(instance._getOb(analysis_id))

        # Does the analysis exists in descendants?
        from_descendant = self.get_from_descendant(instance, service)
        analyses.extend(from_descendant)
        return analyses

    def get_analyses_from_descendants(self, instance):
        """Returns all the analyses from descendants
        """
        analyses = []
        for descendant in instance.getDescendants(all_descendants=True):
            analyses.extend(descendant.objectValues("Analysis"))
        return analyses

    def get_from_instance(self, instance, service):
        """Returns analyses for the given service from the instance
        """
        service_uid = api.get_uid(service)
        analyses = instance.objectValues("Analysis")
        # Filter those analyses with same keyword. Note that a Sample can
        # contain more than one analysis with same keyword because of retests
        return filter(lambda an: an.getServiceUID() == service_uid, analyses)

    def get_from_ancestor(self, instance, service):
        """Returns analyses for the given service from ancestors
        """
        ancestor = instance.getParentAnalysisRequest()
        if not ancestor:
            return []

        analyses = self.get_from_instance(ancestor, service)
        return analyses or self.get_from_ancestor(ancestor, service)

    def get_from_descendant(self, instance, service):
        """Returns analyses for the given service from descendants
        """
        analyses = []
        for descendant in instance.getDescendants():
            # Does the analysis exists in the current descendant?
            descendant_analyses = self.get_from_instance(descendant, service)
            if descendant_analyses:
                analyses.extend(descendant_analyses)

            # Search in descendants from current descendant
            from_descendant = self.get_from_descendant(descendant, service)
            analyses.extend(from_descendant)

        return analyses

    def _to_service(self, thing):
        """Convert to Analysis Service

        :param thing: UID/Catalog Brain/Object/Something
        :returns: Analysis Service object or None
        """

        # Convert UIDs to objects
        if api.is_uid(thing):
            thing = api.get_object_by_uid(thing, None)

        # Bail out if the thing is not a valid object
        if not api.is_object(thing):
            logger.warn("'{}' is not a valid object!".format(repr(thing)))
            return None

        # Ensure we have an object here and not a brain
        obj = api.get_object(thing)

        if IAnalysisService.providedBy(obj):
            return obj

        if IAnalysis.providedBy(obj):
            return obj.getAnalysisService()

        # An object, but neither an Analysis nor AnalysisService?
        # This should never happen.
        portal_type = api.get_portal_type(obj)
        logger.error("ARAnalysesField doesn't accept objects from {} type. "
                     "The object will be dismissed.".format(portal_type))
        return None


registerField(ARAnalysesField,
              title="Analyses",
              description="Manages Analyses of ARs")


1		# -- coding: utf-8 --
2		#
3		# This file is part of SENAITE.CORE.
4		#
5		# SENAITE.CORE is free software: you can redistribute it and/or modify it under
6		# the terms of the GNU General Public License as published by the Free Software
7		# Foundation, version 2.
8		#
9		# This program is distributed in the hope that it will be useful, but WITHOUT
10		# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
11		# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
12		# details.
13		#
14		# You should have received a copy of the GNU General Public License along with
15		# this program; if not, write to the Free Software Foundation, Inc., 51
16		# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17		#
18		# Copyright 2018-2019 by it's authors.
19		# Some rights reserved, see README and LICENSE.
20
21		import itertools
22
23		from AccessControl import ClassSecurityInfo
24		from AccessControl import Unauthorized
25		from Products.Archetypes.Registry import registerField
26		from Products.Archetypes.public import Field
27		from Products.Archetypes.public import ObjectField
28		from zope.interface import alsoProvides
29		from zope.interface import implements
30		from zope.interface import noLongerProvides
31
32		from bika.lims import api
33		from bika.lims import logger
34		from bika.lims.api.security import check_permission
35		from bika.lims.catalog import CATALOG_ANALYSIS_LISTING
36		from bika.lims.catalog import SETUP_CATALOG
37		from bika.lims.interfaces import IARAnalysesField
38		from bika.lims.interfaces import IAnalysis
39		from bika.lims.interfaces import IAnalysisService
40		from bika.lims.interfaces import IInternalUse
41		from bika.lims.interfaces import ISubmitted
42		from bika.lims.permissions import AddAnalysis
43		from bika.lims.utils.analysis import create_analysis
44
45		"""Field to manage Analyses on ARs
46
47		Please see the assigned doctest at tests/doctests/ARAnalysesField.rst
48
49		Run this test from the buildout directory:
50
51		bin/test test_textual_doctests -t ARAnalysesField
52		"""
53
54
55		class ARAnalysesField(ObjectField):
56		"""A field that stores Analyses instances
57		"""
58		implements(IARAnalysesField)
59
60		security = ClassSecurityInfo()
61		_properties = Field._properties.copy()
62		_properties.update({
63		"type": "analyses",
64		"default": None,
65		})
66
67		security.declarePrivate('get')
68
69		def get(self, instance, **kwargs):
70		"""Returns a list of Analyses assigned to this AR
71
72		Return a list of catalog brains unless `full_objects=True` is passed.
73		Other keyword arguments are passed to bika_analysis_catalog
74
75		:param instance: Analysis Request object
76		:param kwargs: Keyword arguments to inject in the search query
77		:returns: A list of Analysis Objects/Catalog Brains
78		"""
79		# Do we need to return objects or brains
80		full_objects = kwargs.get("full_objects", False)
81
82		# Bail out parameters from kwargs that don't match with indexes
83		catalog = api.get_tool(CATALOG_ANALYSIS_LISTING)
84		indexes = catalog.indexes()
85		query = dict([(k, v) for k, v in kwargs.items() if k in indexes])
86
87		# Do the search against the catalog
88		query["portal_type"] = "Analysis"
89		query["getAncestorsUIDs"] = api.get_uid(instance)
90		brains = catalog(query)
91		if full_objects:
92		return map(api.get_object, brains)
93		return brains
94
95		security.declarePrivate('set')
96
97		def set(self, instance, items, prices=None, specs=None, hidden=None, **kw):
98		"""Set/Assign Analyses to this AR
99
100		:param items: List of Analysis objects/brains, AnalysisService
101		objects/brains and/or Analysis Service uids
102		:type items: list
103		:param prices: Mapping of AnalysisService UID -> price
104		:type prices: dict
105		:param specs: List of AnalysisService UID -> Result Range mappings
106		:type specs: list
107		:param hidden: List of AnalysisService UID -> Hidden mappings
108		:type hidden: list
109		:returns: list of new assigned Analyses
110		"""
111		if items is None:
112		items = []
113
114		# Bail out if the items is not a list type
115		if not isinstance(items, (list, tuple)):
116		raise TypeError(
117		"Items parameter must be a tuple or list, got '{}'".format(
118		type(items)))
119
120		# Bail out if the AR is inactive
121		if not api.is_active(instance):
122		raise Unauthorized("Inactive ARs can not be modified"
123		.format(AddAnalysis))
124
125		# Bail out if the user has not the right permission
126		if not check_permission(AddAnalysis, instance):
127		raise Unauthorized("You do not have the '{}' permission"
128		.format(AddAnalysis))
129
130		# Convert the items to a valid list of AnalysisServices
131		services = filter(None, map(self._to_service, items))
132
133		# Calculate dependencies
134		dependencies = map(lambda s: s.getServiceDependencies(), services)
135		dependencies = list(itertools.chain.from_iterable(dependencies))
136
137		# Merge dependencies and services
138		services = set(services + dependencies)
139
140		# Modify existing AR specs with new form values of selected analyses
141		specs = self.resolve_specs(instance, specs)
142
143		# Add analyses
144		params = dict(prices=prices, hidden=hidden, specs=specs)
145		map(lambda serv: self.add_analysis(instance, serv, **params), services)
146
147		# Get all analyses (those from descendants included)
148		analyses = instance.objectValues("Analysis")
149		analyses.extend(self.get_analyses_from_descendants(instance))
150
151		# Bail out those not in services list or submitted
152		uids = map(api.get_uid, services)
153		to_remove = filter(lambda an: an.getServiceUID() not in uids, analyses)
154		to_remove = filter(lambda an: not ISubmitted.providedBy(an), to_remove)
155
156		# Remove analyses
157		map(self.remove_analysis, to_remove)
158
159		def resolve_specs(self, instance, results_ranges):
160		"""Returns a dictionary where the key is the service_uid and the value
161		is its results range. The dictionary is made by extending the
162		results_ranges passed-in with the Sample's ResultsRanges (a copy of the
163		specifications initially set)
164		"""
165		rrs = results_ranges or []
166
167		# Sample's Results ranges
168		sample_rrs = instance.getResultsRange()
169
170		# Ensure all subfields from specification are kept and missing values
171		# for subfields are filled in accordance with the specs
172		rrs = map(lambda rr: self.resolve_range(rr, sample_rrs), rrs)
173
174		# Append those from sample that are missing in the ranges passed-in
175		service_uids = map(lambda rr: rr["uid"], rrs)
176		rrs.extend(filter(lambda rr: rr["uid"] not in service_uids, sample_rrs))
177
178		# Create a dict for easy access to results ranges
179		return dict(map(lambda rr: (rr["uid"], rr), rrs))
180
181		def resolve_range(self, result_range, sample_result_ranges):
182		"""Resolves the range by adding the uid if not present and filling the
183		missing subfield values with those that come from the Sample
184		specification if they are not present in the result_range passed-in
185		"""
186		# Resolve result_range to make sure it contain uid subfield
187		rrs = self.resolve_uid(result_range)
188		uid = rrs.get("uid")
189
190		for sample_rr in sample_result_ranges:
191		if uid and sample_rr.get("uid") == uid:
192		# Keep same fields from sample
193		rr = sample_rr.copy()
194		rr.update(rrs)
195		return rr
196
197		# Return the original with no changes
198		return rrs
199
200	View Code Duplication	def resolve_uid(self, result_range):
		0 ignored issues – show Duplication introduced 2020-01-23 19:50 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
201		"""Resolves the uid key for the result_range passed in if it does not
202		exist when contains a keyword
203		"""
204		value = result_range.copy()
205		uid = value.get("uid")
206		if api.is_uid(uid) and uid != "0":
207		return value
208
209		# uid key does not exist or is not valid, try to infere from keyword
210		keyword = value.get("keyword")
211		if keyword:
212		query = dict(portal_type="AnalysisService", getKeyword=keyword)
213		brains = api.search(query, SETUP_CATALOG)
214		if len(brains) == 1:
215		uid = api.get_uid(brains[0])
216		value["uid"] = uid
217		return value
218
219		def add_analysis(self, instance, service, **kwargs):
220		service_uid = api.get_uid(service)
221
222		# Ensure we have suitable parameters
223		specs = kwargs.get("specs") or {}
224
225		# Get the hidden status for the service
226		hidden = kwargs.get("hidden") or []
227		hidden = filter(lambda d: d.get("uid") == service_uid, hidden)
228		hidden = hidden and hidden[0].get("hidden") or service.getHidden()
229
230		# Get the price for the service
231		prices = kwargs.get("prices") or {}
232		price = prices.get(service_uid) or service.getPrice()
233
234		# Gets the analysis or creates the analysis for this service
235		# Note this returns a list, because is possible to have multiple
236		# partitions with same analysis
237		analyses = self.resolve_analyses(instance, service)
238		if not analyses:
239		# Create the analysis
240		keyword = service.getKeyword()
241		logger.info("Creating new analysis '{}'".format(keyword))
242		analysis = create_analysis(instance, service)
243		analyses.append(analysis)
244
245		skip = ["cancelled", "retracted", "rejected"]
246		for analysis in analyses:
247		# Skip analyses to better not modify
248		if api.get_review_status(analysis) in skip:
249		continue
250
251		# Set the hidden status
252		analysis.setHidden(hidden)
253
254		# Set the price of the Analysis
255		analysis.setPrice(price)
256
257		# Set the internal use status
258		parent_sample = analysis.getRequest()
259		analysis.setInternalUse(parent_sample.getInternalUse())
260
261		# Set the result range to the analysis
262		analysis_rr = specs.get(service_uid) or analysis.getResultsRange()
263		analysis.setResultsRange(analysis_rr)
264		analysis.reindexObject()
265
266		def remove_analysis(self, analysis):
267		"""Removes a given analysis from the instance
268		"""
269		# Remember assigned attachments
270		# https://github.com/senaite/senaite.core/issues/1025
271		attachments = analysis.getAttachment()
272		analysis.setAttachment([])
273
274		# If assigned to a worksheet, unassign it before deletion
275		worksheet = analysis.getWorksheet()
276		if worksheet:
277		worksheet.removeAnalysis(analysis)
278
279		# Remove the analysis
280		# Note the analysis might belong to a partition
281		analysis.aq_parent.manage_delObjects(ids=[api.get_id(analysis)])
282
283		# Remove orphaned attachments
284		for attachment in attachments:
285		if not attachment.getLinkedAnalyses():
286		# only delete attachments which are no further linked
287		logger.info(
288		"Deleting attachment: {}".format(attachment.getId()))
289		attachment_id = api.get_id(attachment)
290		api.get_parent(attachment).manage_delObjects(attachment_id)
291
292		def resolve_analyses(self, instance, service):
293		"""Resolves analyses for the service and instance
294		It returns a list, cause for a given sample, multiple analyses for same
295		service can exist due to the possibility of having multiple partitions
296		"""
297		analyses = []
298
299		# Does the analysis exists in this instance already?
300		instance_analyses = self.get_from_instance(instance, service)
301		if instance_analyses:
302		analyses.extend(instance_analyses)
303
304		# Does the analysis exists in an ancestor?
305		from_ancestor = self.get_from_ancestor(instance, service)
306		for ancestor_analysis in from_ancestor:
307		# Move the analysis into this instance. The ancestor's
308		# analysis will be masked otherwise
309		analysis_id = api.get_id(ancestor_analysis)
310		logger.info("Analysis {} is from an ancestor".format(analysis_id))
311		cp = ancestor_analysis.aq_parent.manage_cutObjects(analysis_id)
312		instance.manage_pasteObjects(cp)
313		analyses.append(instance._getOb(analysis_id))
314
315		# Does the analysis exists in descendants?
316		from_descendant = self.get_from_descendant(instance, service)
317		analyses.extend(from_descendant)
318		return analyses
319
320		def get_analyses_from_descendants(self, instance):
321		"""Returns all the analyses from descendants
322		"""
323		analyses = []
324		for descendant in instance.getDescendants(all_descendants=True):
325		analyses.extend(descendant.objectValues("Analysis"))
326		return analyses
327
328		def get_from_instance(self, instance, service):
329		"""Returns analyses for the given service from the instance
330		"""
331		service_uid = api.get_uid(service)
332		analyses = instance.objectValues("Analysis")
333		# Filter those analyses with same keyword. Note that a Sample can
334		# contain more than one analysis with same keyword because of retests
335		return filter(lambda an: an.getServiceUID() == service_uid, analyses)
336
337		def get_from_ancestor(self, instance, service):
338		"""Returns analyses for the given service from ancestors
339		"""
340		ancestor = instance.getParentAnalysisRequest()
341		if not ancestor:
342		return []
343
344		analyses = self.get_from_instance(ancestor, service)
345		return analyses or self.get_from_ancestor(ancestor, service)
346
347		def get_from_descendant(self, instance, service):
348		"""Returns analyses for the given service from descendants
349		"""
350		analyses = []
351		for descendant in instance.getDescendants():
352		# Does the analysis exists in the current descendant?
353		descendant_analyses = self.get_from_instance(descendant, service)
354		if descendant_analyses:
355		analyses.extend(descendant_analyses)
356
357		# Search in descendants from current descendant
358		from_descendant = self.get_from_descendant(descendant, service)
359		analyses.extend(from_descendant)
360
361		return analyses
362
363		def _to_service(self, thing):
364		"""Convert to Analysis Service
365
366		:param thing: UID/Catalog Brain/Object/Something
367		:returns: Analysis Service object or None
368		"""
369
370		# Convert UIDs to objects
371		if api.is_uid(thing):
372		thing = api.get_object_by_uid(thing, None)
373
374		# Bail out if the thing is not a valid object
375		if not api.is_object(thing):
376		logger.warn("'{}' is not a valid object!".format(repr(thing)))
377		return None
378
379		# Ensure we have an object here and not a brain
380		obj = api.get_object(thing)
381
382		if IAnalysisService.providedBy(obj):
383		return obj
384
385		if IAnalysis.providedBy(obj):
386		return obj.getAnalysisService()
387
388		# An object, but neither an Analysis nor AnalysisService?
389		# This should never happen.
390		portal_type = api.get_portal_type(obj)
391		logger.error("ARAnalysesField doesn't accept objects from {} type. "
392		"The object will be dismissed.".format(portal_type))
393		return None
394
395
396		registerField(ARAnalysesField,
397		title="Analyses",
398		description="Manages Analyses of ARs")
399

senaite / senaite.core

Push — master ( 2d15ca...15a906 )

ARAnalysesField.get_from_descendant() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like