bika.lims.browser.fields.aranalysesfield.ARAnalysesField.set() - Code Metrics - Inspection of "Refactored and fixed inconsistencies with Analysis..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( caa188...9d0ad3 )

by Ramon

created 2018-10-01 19:11 UTC

ARAnalysesField.set() F

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

Complexity

Conditions

Size

Total Lines	125
Code Lines	58

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	58
dl	0
loc	125
rs	2.4
c	0
b	0
f	0
cc	16
nop	6

How to fix Long Method Complexity

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

If many parameters/temporary variables are present:
- Replace temporary variables with Query
- Introduce parameter object; often combined with preserve whole object
- If the above two are insufficient: Replace method with method object
If you have long conditionals: Decompose Conditional
Otherwise: Extract method

Complexity

Complex classes like bika.lims.browser.fields.aranalysesfield.ARAnalysesField.set() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE
#
# Copyright 2018 by it's authors.
# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.

import itertools

from AccessControl import ClassSecurityInfo
from bika.lims import api, deprecated, logger
from bika.lims.catalog import CATALOG_ANALYSIS_LISTING
from bika.lims.interfaces import IAnalysis, IAnalysisService, IARAnalysesField
from bika.lims.permissions import ViewRetractedAnalyses
from bika.lims.utils.analysis import create_analysis
from bika.lims.workflow import getReviewHistoryActionsList
from Products.Archetypes.public import Field, ObjectField
from Products.Archetypes.Registry import registerField
from Products.Archetypes.utils import shasattr
from Products.CMFCore.utils import getToolByName
from zope.interface import implements

"""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.
        Overrides "ViewRetractedAnalyses" when `retracted=True` is passed.
        Other keyword arguments are passed to bika_analysis_catalog

        :param instance: Analysis Request object
        :param kwargs: Keyword arguments to be passed to control the output
        :returns: A list of Analysis Objects/Catalog Brains
        """

        full_objects = False
        # If reflexed is false don't return the analyses that have been
        # reflexed, only the final ones
        reflexed = True

        if 'full_objects' in kwargs:
            full_objects = kwargs['full_objects']
            del kwargs['full_objects']

        if 'reflexed' in kwargs:
            reflexed = kwargs['reflexed']
            del kwargs['reflexed']

        if 'retracted' in kwargs:
            retracted = kwargs['retracted']
            del kwargs['retracted']
        else:
            mtool = getToolByName(instance, 'portal_membership')
            retracted = mtool.checkPermission(
                ViewRetractedAnalyses, instance)

        bac = getToolByName(instance, CATALOG_ANALYSIS_LISTING)
        contentFilter = dict([(k, v) for k, v in kwargs.items()
                              if k in bac.indexes()])
        contentFilter['portal_type'] = "Analysis"
        contentFilter['sort_on'] = "getKeyword"
        contentFilter['path'] = {'query': api.get_path(instance),
                                 'level': 0}
        analyses = bac(contentFilter)
        if not retracted or full_objects or not reflexed:
            analyses_filtered = []
            for a in analyses:
                if not retracted and a.review_state == 'retracted':
                    continue
                if full_objects or not reflexed:
                    a_obj = a.getObject()
                    # Check if analysis has been reflexed
                    if not reflexed and \
                            a_obj.getReflexRuleActionsTriggered() != '':
                        continue
                    if full_objects:
                        a = a_obj
                analyses_filtered.append(a)
            analyses = analyses_filtered
        return analyses

    security.declarePrivate('set')

    def set(self, instance, items, prices=None, specs=None, **kwargs):
        """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 Record mappings
        :type specs: list
        :returns: list of new assigned Analyses
        """

        # This setter returns a list of new set Analyses
        new_analyses = []

        # Prevent removing all Analyses
        if not items:
            logger.warn("Not allowed to remove all Analyses from AR.")
            return new_analyses

        # 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 in frozen state
        if self._is_frozen(instance):
            raise ValueError(
                "Analyses can not be modified for inactive/verified ARs")

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

        # Calculate dependencies
        # FIXME Infinite recursion error possible here, if the formula includes
        #       the Keyword of the Service that includes the Calculation
        dependencies = map(lambda s: s.getServiceDependencies(), services)
        dependencies = list(itertools.chain.from_iterable(dependencies))

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

        # Service UIDs
        service_uids = map(api.get_uid, services)

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

        for service in services:
            keyword = service.getKeyword()

            # Create the Analysis if it doesn't exist
            if shasattr(instance, keyword):
                analysis = instance._getOb(keyword)
            else:
                # TODO Entry point for interims assignment and Calculation
                #      decoupling from Analysis. See comments PR#593
                analysis = create_analysis(instance, service)
                # TODO Remove when the `create_analysis` function supports this
                # Set the interim fields only for new created Analysis
                self._update_interims(analysis, service)
                new_analyses.append(analysis)

            # Set the price of the Analysis
            self._update_price(analysis, service, prices)

        # delete analyses
        delete_ids = []
        assigned_attachments = []

        for analysis in instance.objectValues('Analysis'):
            service_uid = analysis.getServiceUID()

            # Skip assigned Analyses
            if service_uid in service_uids:
                continue

            # Skip Analyses in frozen states
            if self._is_frozen(analysis, "retract"):
                logger.warn("Inactive/verified/retracted Analyses can not be "
                            "removed.")
                continue

            # Remember assigned attachments
            # https://github.com/senaite/senaite.core/issues/1025
            assigned_attachments.extend(analysis.getAttachment())
            analysis.setAttachment([])

            # If it is assigned to a worksheet, unassign it before deletion.
            if self._is_assigned_to_worksheet(analysis):
                backrefs = self._get_assigned_worksheets(analysis)
                ws = backrefs[0]
                ws.removeAnalysis(analysis)

            # Unset the partition reference
            part = analysis.getSamplePartition()
            if part:
                # From this partition, remove the reference to the current
                # analysis that is going to be removed to prevent inconsistent
                # states (Sample Partitions referencing to Analyses that do not
                # exist anymore
                an_uid = api.get_uid(analysis)
                part_ans = part.getAnalyses() or []
                part_ans = filter(lambda an: api.get_uid(an) != an_uid, part_ans)

                part.setAnalyses(part_ans)
            # Unset the Analysis-to-Partition reference
            analysis.setSamplePartition(None)
            delete_ids.append(analysis.getId())

        if delete_ids:
            # Note: subscriber might promote the AR
            instance.manage_delObjects(ids=delete_ids)

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

        return new_analyses

    def _get_services(self, full_objects=False):
        """Fetch and return analysis service objects
        """
        bsc = api.get_tool('bika_setup_catalog')
        brains = bsc(portal_type='AnalysisService')
        if full_objects:
            return map(api.get_object, brains)
        return brains

    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.
        msg = "ARAnalysesField doesn't accept objects from {} type. " \
            "The object will be dismissed.".format(api.get_portal_type(obj))
        logger.warn(msg)
        return None

    def _is_frozen(self, brain_or_object, *frozen_transitions):
        """Check if the passed in object is frozen: the object is cancelled,
        inactive or has been verified at some point
        :param brain_or_object: Analysis or AR Brain/Object
        :param frozen_transitions: additional transitions that freeze the object
        :returns: True if the object is frozen
        """
        if not api.is_active(brain_or_object):
            return True
        object = api.get_object(brain_or_object)
        frozen_trans = set(frozen_transitions)
        frozen_trans.add('verify')
        performed_transitions = set(getReviewHistoryActionsList(object))
        if frozen_trans.intersection(performed_transitions):
            return True
        return False

    def _get_assigned_worksheets(self, analysis):
        """Return the assigned worksheets of this Analysis

        :param analysis: Analysis Brain/Object
        :returns: Worksheet Backreferences
        """
        analysis = api.get_object(analysis)
        return analysis.getBackReferences("WorksheetAnalysis")

    def _is_assigned_to_worksheet(self, analysis):
        """Check if the Analysis is assigned to a worksheet

        :param analysis: Analysis Brain/Object
        :returns: True if the Analysis is assigned to a WS
        """
        analysis = api.get_object(analysis)
        state = api.get_workflow_status_of(
            analysis, state_var='worksheetanalysis_review_state')
        return state == "assigned"

    def _update_interims(self, analysis, service):
        """Update Interim Fields of the Analysis

        :param analysis: Analysis Object
        :param service: Analysis Service Object
        """
        service_interims = service.getInterimFields()
        analysis.setInterimFields(service_interims)

    def _update_price(self, analysis, service, prices):
        """Update the Price of the Analysis

        :param analysis: Analysis Object
        :param service: Analysis Service Object
        :param prices: Price mapping
        """
        prices = prices or {}
        price = prices.get(service.UID(), service.getPrice())
        analysis.setPrice(price)

    def _update_specs(self, instance, specs):
        """Update AR specifications

        :param instance: Analysis Request
        :param specs: List of Specification Records
        """

        if specs is None:
            return

        # N.B. we copy the records here, otherwise the spec will be written to
        #      the attached specification of this AR
        rr = {item["keyword"]: item.copy()
              for item in instance.getResultsRange()}
        for spec in specs:
            keyword = spec.get("keyword")
            if keyword in rr:
                # overwrite the instance specification only, if the specific
                # analysis spec has min/max values set
                if all([spec.get("min"), spec.get("max")]):
                    rr[keyword].update(spec)
                else:
                    rr[keyword] = spec
            else:
                rr[keyword] = spec
        return instance.setResultsRange(rr.values())

    # DEPRECATED: The following code should not be in the field's domain

    security.declarePublic('Vocabulary')

    @deprecated("Please refactor, this method will be removed in senaite.core 1.5")
    def Vocabulary(self, content_instance=None):
        """Create a vocabulary from analysis services
        """
        vocab = []
        for service in self._get_services():
            vocab.append((api.get_uid(service), api.get_title(service)))
        return vocab

    security.declarePublic('Services')

    @deprecated("Please refactor, this method will be removed in senaite.core 1.5")
    def Services(self):
        """Fetch and return analysis service objects
        """
        return self._get_services(full_objects=True)


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


1			# -- coding: utf-8 --
2			#
3			# This file is part of SENAITE.CORE
4			#
5			# Copyright 2018 by it's authors.
6			# Some rights reserved. See LICENSE.rst, CONTRIBUTORS.rst.
7
8			import itertools
9
10			from AccessControl import ClassSecurityInfo
11			from bika.lims import api, deprecated, logger
12			from bika.lims.catalog import CATALOG_ANALYSIS_LISTING
13			from bika.lims.interfaces import IAnalysis, IAnalysisService, IARAnalysesField
14			from bika.lims.permissions import ViewRetractedAnalyses
15			from bika.lims.utils.analysis import create_analysis
16			from bika.lims.workflow import getReviewHistoryActionsList
17			from Products.Archetypes.public import Field, ObjectField
18			from Products.Archetypes.Registry import registerField
19			from Products.Archetypes.utils import shasattr
20			from Products.CMFCore.utils import getToolByName
21			from zope.interface import implements
22
23			"""Field to manage Analyses on ARs
24
25			Please see the assigned doctest at tests/doctests/ARAnalysesField.rst
26
27			Run this test from the buildout directory:
28
29			bin/test test_textual_doctests -t ARAnalysesField
30			"""
31
32
33			class ARAnalysesField(ObjectField):
34			"""A field that stores Analyses instances
35			"""
36			implements(IARAnalysesField)
37
38			security = ClassSecurityInfo()
39			_properties = Field._properties.copy()
40			_properties.update({
41			'type': 'analyses',
42			'default': None,
43			})
44
45			security.declarePrivate('get')
46
47			def get(self, instance, **kwargs):
48			"""Returns a list of Analyses assigned to this AR
49
50			Return a list of catalog brains unless `full_objects=True` is passed.
51			Overrides "ViewRetractedAnalyses" when `retracted=True` is passed.
52			Other keyword arguments are passed to bika_analysis_catalog
53
54			:param instance: Analysis Request object
55			:param kwargs: Keyword arguments to be passed to control the output
56			:returns: A list of Analysis Objects/Catalog Brains
57			"""
58
59			full_objects = False
60			# If reflexed is false don't return the analyses that have been
61			# reflexed, only the final ones
62			reflexed = True
63
64			if 'full_objects' in kwargs:
65			full_objects = kwargs['full_objects']
66			del kwargs['full_objects']
67
68			if 'reflexed' in kwargs:
69			reflexed = kwargs['reflexed']
70			del kwargs['reflexed']
71
72			if 'retracted' in kwargs:
73			retracted = kwargs['retracted']
74			del kwargs['retracted']
75			else:
76			mtool = getToolByName(instance, 'portal_membership')
77			retracted = mtool.checkPermission(
78			ViewRetractedAnalyses, instance)
79
80			bac = getToolByName(instance, CATALOG_ANALYSIS_LISTING)
81			contentFilter = dict([(k, v) for k, v in kwargs.items()
82			if k in bac.indexes()])
83			contentFilter['portal_type'] = "Analysis"
84			contentFilter['sort_on'] = "getKeyword"
85			contentFilter['path'] = {'query': api.get_path(instance),
86			'level': 0}
87			analyses = bac(contentFilter)
88			if not retracted or full_objects or not reflexed:
89			analyses_filtered = []
90			for a in analyses:
91			if not retracted and a.review_state == 'retracted':
92			continue
93			if full_objects or not reflexed:
94			a_obj = a.getObject()
95			# Check if analysis has been reflexed
96			if not reflexed and \
97			a_obj.getReflexRuleActionsTriggered() != '':
98			continue
99			if full_objects:
100			a = a_obj
101			analyses_filtered.append(a)
102			analyses = analyses_filtered
103			return analyses
104
105			security.declarePrivate('set')
106
107			def set(self, instance, items, prices=None, specs=None, **kwargs):
108			"""Set/Assign Analyses to this AR
109
110			:param items: List of Analysis objects/brains, AnalysisService
111			objects/brains and/or Analysis Service uids
112			:type items: list
113			:param prices: Mapping of AnalysisService UID -> price
114			:type prices: dict
115			:param specs: List of AnalysisService UID -> Result Range Record mappings
116			:type specs: list
117			:returns: list of new assigned Analyses
118			"""
119
120			# This setter returns a list of new set Analyses
121			new_analyses = []
122
123			# Prevent removing all Analyses
124			if not items:
125			logger.warn("Not allowed to remove all Analyses from AR.")
126			return new_analyses
127
128			# Bail out if the items is not a list type
129			if not isinstance(items, (list, tuple)):
130			raise TypeError(
131			"Items parameter must be a tuple or list, got '{}'".format(
132			type(items)))
133
134			# Bail out if the AR in frozen state
135			if self._is_frozen(instance):
136			raise ValueError(
137			"Analyses can not be modified for inactive/verified ARs")
138
139			# Convert the items to a valid list of AnalysisServices
140			services = filter(None, map(self._to_service, items))
141
142			# Calculate dependencies
143			# FIXME Infinite recursion error possible here, if the formula includes
144			# the Keyword of the Service that includes the Calculation
145			dependencies = map(lambda s: s.getServiceDependencies(), services)
146			dependencies = list(itertools.chain.from_iterable(dependencies))
147
148			# Merge dependencies and services
149			services = set(services + dependencies)
150
151			# Service UIDs
152			service_uids = map(api.get_uid, services)
153
154			# Modify existing AR specs with new form values of selected analyses.
155			self._update_specs(instance, specs)
156
157			for service in services:
158			keyword = service.getKeyword()
159
160			# Create the Analysis if it doesn't exist
161			if shasattr(instance, keyword):
162			analysis = instance._getOb(keyword)
163			else:
164			# TODO Entry point for interims assignment and Calculation
165			# decoupling from Analysis. See comments PR#593
166			analysis = create_analysis(instance, service)
167			# TODO Remove when the `create_analysis` function supports this
168			# Set the interim fields only for new created Analysis
169			self._update_interims(analysis, service)
170			new_analyses.append(analysis)
171
172			# Set the price of the Analysis
173			self._update_price(analysis, service, prices)
174
175			# delete analyses
176			delete_ids = []
177			assigned_attachments = []
178
179			for analysis in instance.objectValues('Analysis'):
180			service_uid = analysis.getServiceUID()
181
182			# Skip assigned Analyses
183			if service_uid in service_uids:
184			continue
185
186			# Skip Analyses in frozen states
187			if self._is_frozen(analysis, "retract"):
188			logger.warn("Inactive/verified/retracted Analyses can not be "
189			"removed.")
190			continue
191
192			# Remember assigned attachments
193			# https://github.com/senaite/senaite.core/issues/1025
194			assigned_attachments.extend(analysis.getAttachment())
195			analysis.setAttachment([])
196
197			# If it is assigned to a worksheet, unassign it before deletion.
198			if self._is_assigned_to_worksheet(analysis):
199			backrefs = self._get_assigned_worksheets(analysis)
200			ws = backrefs[0]
201			ws.removeAnalysis(analysis)
202
203			# Unset the partition reference
204			part = analysis.getSamplePartition()
205			if part:
206			# From this partition, remove the reference to the current
207			# analysis that is going to be removed to prevent inconsistent
208			# states (Sample Partitions referencing to Analyses that do not
209			# exist anymore
210			an_uid = api.get_uid(analysis)
211			part_ans = part.getAnalyses() or []
212			part_ans = filter(lambda an: api.get_uid(an) != an_uid, part_ans)
			0 ignored issues – show introduced 2018-07-25 08:58 UTC by Report Bug Copy Issue Report The variable `an_uid` does not seem to be defined for all execution paths. Loading history...
213			part.setAnalyses(part_ans)
214			# Unset the Analysis-to-Partition reference
215			analysis.setSamplePartition(None)
216			delete_ids.append(analysis.getId())
217
218			if delete_ids:
219			# Note: subscriber might promote the AR
220			instance.manage_delObjects(ids=delete_ids)
221
222			# Remove orphaned attachments
223			for attachment in assigned_attachments:
224			# only delete attachments which are no further linked
225			if not attachment.getLinkedAnalyses():
226			logger.info(
227			"Deleting attachment: {}".format(attachment.getId()))
228			attachment_id = api.get_id(attachment)
229			api.get_parent(attachment).manage_delObjects(attachment_id)
230
231			return new_analyses
232
233			def _get_services(self, full_objects=False):
234			"""Fetch and return analysis service objects
235			"""
236			bsc = api.get_tool('bika_setup_catalog')
237			brains = bsc(portal_type='AnalysisService')
238			if full_objects:
239			return map(api.get_object, brains)
240			return brains
241
242			def _to_service(self, thing):
243			"""Convert to Analysis Service
244
245			:param thing: UID/Catalog Brain/Object/Something
246			:returns: Analysis Service object or None
247			"""
248
249			# Convert UIDs to objects
250			if api.is_uid(thing):
251			thing = api.get_object_by_uid(thing, None)
252
253			# Bail out if the thing is not a valid object
254			if not api.is_object(thing):
255			logger.warn("'{}' is not a valid object!".format(repr(thing)))
256			return None
257
258			# Ensure we have an object here and not a brain
259			obj = api.get_object(thing)
260
261			if IAnalysisService.providedBy(obj):
262			return obj
263
264			if IAnalysis.providedBy(obj):
265			return obj.getAnalysisService()
266
267			# An object, but neither an Analysis nor AnalysisService?
268			# This should never happen.
269			msg = "ARAnalysesField doesn't accept objects from {} type. " \
270			"The object will be dismissed.".format(api.get_portal_type(obj))
271			logger.warn(msg)
272			return None
273
274			def _is_frozen(self, brain_or_object, *frozen_transitions):
275			"""Check if the passed in object is frozen: the object is cancelled,
276			inactive or has been verified at some point
277			:param brain_or_object: Analysis or AR Brain/Object
278			:param frozen_transitions: additional transitions that freeze the object
279			:returns: True if the object is frozen
280			"""
281			if not api.is_active(brain_or_object):
282			return True
283			object = api.get_object(brain_or_object)
284			frozen_trans = set(frozen_transitions)
285			frozen_trans.add('verify')
286			performed_transitions = set(getReviewHistoryActionsList(object))
287			if frozen_trans.intersection(performed_transitions):
288			return True
289			return False
290
291			def _get_assigned_worksheets(self, analysis):
292			"""Return the assigned worksheets of this Analysis
293
294			:param analysis: Analysis Brain/Object
295			:returns: Worksheet Backreferences
296			"""
297			analysis = api.get_object(analysis)
298			return analysis.getBackReferences("WorksheetAnalysis")
299
300			def _is_assigned_to_worksheet(self, analysis):
301			"""Check if the Analysis is assigned to a worksheet
302
303			:param analysis: Analysis Brain/Object
304			:returns: True if the Analysis is assigned to a WS
305			"""
306			analysis = api.get_object(analysis)
307			state = api.get_workflow_status_of(
308			analysis, state_var='worksheetanalysis_review_state')
309			return state == "assigned"
310
311			def _update_interims(self, analysis, service):
312			"""Update Interim Fields of the Analysis
313
314			:param analysis: Analysis Object
315			:param service: Analysis Service Object
316			"""
317			service_interims = service.getInterimFields()
318			analysis.setInterimFields(service_interims)
319
320			def _update_price(self, analysis, service, prices):
321			"""Update the Price of the Analysis
322
323			:param analysis: Analysis Object
324			:param service: Analysis Service Object
325			:param prices: Price mapping
326			"""
327			prices = prices or {}
328			price = prices.get(service.UID(), service.getPrice())
329			analysis.setPrice(price)
330
331			def _update_specs(self, instance, specs):
332			"""Update AR specifications
333
334			:param instance: Analysis Request
335			:param specs: List of Specification Records
336			"""
337
338			if specs is None:
339			return
340
341			# N.B. we copy the records here, otherwise the spec will be written to
342			# the attached specification of this AR
343			rr = {item["keyword"]: item.copy()
344			for item in instance.getResultsRange()}
345			for spec in specs:
346			keyword = spec.get("keyword")
347			if keyword in rr:
348			# overwrite the instance specification only, if the specific
349			# analysis spec has min/max values set
350			if all([spec.get("min"), spec.get("max")]):
351			rr[keyword].update(spec)
352			else:
353			rr[keyword] = spec
354			else:
355			rr[keyword] = spec
356			return instance.setResultsRange(rr.values())
357
358			# DEPRECATED: The following code should not be in the field's domain
359
360			security.declarePublic('Vocabulary')
361
362			@deprecated("Please refactor, this method will be removed in senaite.core 1.5")
363			def Vocabulary(self, content_instance=None):
364			"""Create a vocabulary from analysis services
365			"""
366			vocab = []
367			for service in self._get_services():
368			vocab.append((api.get_uid(service), api.get_title(service)))
369			return vocab
370
371			security.declarePublic('Services')
372
373			@deprecated("Please refactor, this method will be removed in senaite.core 1.5")
374			def Services(self):
375			"""Fetch and return analysis service objects
376			"""
377			return self._get_services(full_objects=True)
378
379
380			registerField(ARAnalysesField,
381			title="Analyses",
382			description="Manages Analyses of ARs")
383

senaite / senaite.core

Push — master ( caa188...9d0ad3 )

ARAnalysesField.set() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like