build.bika.lims.browser.fields.aranalysesfield.ARAnalysesField.set() - Code Metrics - Inspection of "fixing cobas integra instrument import interface (..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( d8e2ec...90ae0b )

by Jordi

created 2019-03-11 15:48 UTC

ARAnalysesField.set() F

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

Complexity

Conditions

Size

Total Lines	142
Code Lines	64

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	64
dl	0
loc	142
rs	0
c	0
b	0
f	0
cc	20
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 build.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 AccessControl import Unauthorized
from AccessControl import getSecurityManager
from bika.lims import api
from bika.lims import logger
from bika.lims.catalog import CATALOG_ANALYSIS_LISTING
from bika.lims.interfaces import IAnalysis
from bika.lims.interfaces import IAnalysisService
from bika.lims.interfaces import IARAnalysesField
from bika.lims.permissions import AddAnalysis
from bika.lims.utils.analysis import create_analysis
from Products.Archetypes.public import Field
from Products.Archetypes.public import 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.
        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
        """
        catalog = getToolByName(instance, CATALOG_ANALYSIS_LISTING)
        query = dict(
            [(k, v) for k, v in kwargs.items() if k in catalog.indexes()])
        query["portal_type"] = "Analysis"
        query["getRequestUID"] = api.get_uid(instance)
        analyses = catalog(query)
        if not kwargs.get("full_objects", False):
            return analyses

        return map(api.get_object, 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 mappings
        :type specs: list
        :returns: list of new assigned Analyses
        """
        # This setter returns a list of new set Analyses
        new_analyses = []

        # Current assigned analyses
        analyses = instance.objectValues("Analysis")

        # Analyses which are in a non-open state must be retained, except those
        # that are in a registered state (the sample has not been received)
        non_open_analyses = filter(lambda an: not an.isOpen(), analyses)
        non_open_analyses = filter(lambda an: api.get_workflow_status_of(an)
                                              != "registered", non_open_analyses)

        # Prevent removing all analyses
        #
        # N.B.: Non-open analyses are rendered disabled in the HTML form.
        #       Therefore, their UIDs are not included in the submitted UIDs.
        if not items and not non_open_analyses:
            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 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
        sm = getSecurityManager()
        if not sm.checkPermission(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
        # 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)

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

        # CREATE/MODIFY ANALYSES

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

            # Create the Analysis if it doesn't exist
            if shasattr(instance, keyword):
                analysis = instance._getOb(keyword)
            else:
                analysis = create_analysis(instance, service)
                new_analyses.append(analysis)

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

        # DELETE ANALYSES

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

        # Analyses IDs to delete
        delete_ids = []

        # Assigned Attachments
        assigned_attachments = []

        for analysis in analyses:
            service_uid = analysis.getServiceUID()

            # Skip if the Service is selected
            if service_uid in service_uids:
                continue

            # Skip non-open Analyses
            if analysis in non_open_analyses:
                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.
            worksheet = analysis.getWorksheet()
            if worksheet:
                worksheet.removeAnalysis(analysis)

            # Unset the partition reference
            # TODO Remove in >v1.3.0 - This is kept for backwards-compatibility
            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.
        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

    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())


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 AccessControl import Unauthorized
12			from AccessControl import getSecurityManager
13			from bika.lims import api
14			from bika.lims import logger
15			from bika.lims.catalog import CATALOG_ANALYSIS_LISTING
16			from bika.lims.interfaces import IAnalysis
17			from bika.lims.interfaces import IAnalysisService
18			from bika.lims.interfaces import IARAnalysesField
19			from bika.lims.permissions import AddAnalysis
20			from bika.lims.utils.analysis import create_analysis
21			from Products.Archetypes.public import Field
22			from Products.Archetypes.public import ObjectField
23			from Products.Archetypes.Registry import registerField
24			from Products.Archetypes.utils import shasattr
25			from Products.CMFCore.utils import getToolByName
26			from zope.interface import implements
27
28			"""Field to manage Analyses on ARs
29
30			Please see the assigned doctest at tests/doctests/ARAnalysesField.rst
31
32			Run this test from the buildout directory:
33
34			bin/test test_textual_doctests -t ARAnalysesField
35			"""
36
37
38			class ARAnalysesField(ObjectField):
39			"""A field that stores Analyses instances
40			"""
41			implements(IARAnalysesField)
42
43			security = ClassSecurityInfo()
44			_properties = Field._properties.copy()
45			_properties.update({
46			"type": "analyses",
47			"default": None,
48			})
49
50			security.declarePrivate('get')
51
52			def get(self, instance, **kwargs):
53			"""Returns a list of Analyses assigned to this AR
54
55			Return a list of catalog brains unless `full_objects=True` is passed.
56			Other keyword arguments are passed to bika_analysis_catalog
57
58			:param instance: Analysis Request object
59			:param kwargs: Keyword arguments to inject in the search query
60			:returns: A list of Analysis Objects/Catalog Brains
61			"""
62			catalog = getToolByName(instance, CATALOG_ANALYSIS_LISTING)
63			query = dict(
64			[(k, v) for k, v in kwargs.items() if k in catalog.indexes()])
65			query["portal_type"] = "Analysis"
66			query["getRequestUID"] = api.get_uid(instance)
67			analyses = catalog(query)
68			if not kwargs.get("full_objects", False):
69			return analyses
70
71			return map(api.get_object, analyses)
72
73			security.declarePrivate('set')
74
75			def set(self, instance, items, prices=None, specs=None, **kwargs):
76			"""Set/Assign Analyses to this AR
77
78			:param items: List of Analysis objects/brains, AnalysisService
79			objects/brains and/or Analysis Service uids
80			:type items: list
81			:param prices: Mapping of AnalysisService UID -> price
82			:type prices: dict
83			:param specs: List of AnalysisService UID -> Result Range mappings
84			:type specs: list
85			:returns: list of new assigned Analyses
86			"""
87			# This setter returns a list of new set Analyses
88			new_analyses = []
89
90			# Current assigned analyses
91			analyses = instance.objectValues("Analysis")
92
93			# Analyses which are in a non-open state must be retained, except those
94			# that are in a registered state (the sample has not been received)
95			non_open_analyses = filter(lambda an: not an.isOpen(), analyses)
96			non_open_analyses = filter(lambda an: api.get_workflow_status_of(an)
97			!= "registered", non_open_analyses)
98
99			# Prevent removing all analyses
100			#
101			# N.B.: Non-open analyses are rendered disabled in the HTML form.
102			# Therefore, their UIDs are not included in the submitted UIDs.
103			if not items and not non_open_analyses:
104			logger.warn("Not allowed to remove all Analyses from AR.")
105			return new_analyses
106
107			# Bail out if the items is not a list type
108			if not isinstance(items, (list, tuple)):
109			raise TypeError(
110			"Items parameter must be a tuple or list, got '{}'".format(
111			type(items)))
112
113			# Bail out if the AR is inactive
114			if not api.is_active(instance):
115			raise Unauthorized("Inactive ARs can not be modified"
116			.format(AddAnalysis))
117
118			# Bail out if the user has not the right permission
119			sm = getSecurityManager()
120			if not sm.checkPermission(AddAnalysis, instance):
121			raise Unauthorized("You do not have the '{}' permission"
122			.format(AddAnalysis))
123
124			# Convert the items to a valid list of AnalysisServices
125			services = filter(None, map(self._to_service, items))
126
127			# Calculate dependencies
128			# FIXME Infinite recursion error possible here, if the formula includes
129			# the Keyword of the Service that includes the Calculation
130			dependencies = map(lambda s: s.getServiceDependencies(), services)
131			dependencies = list(itertools.chain.from_iterable(dependencies))
132
133			# Merge dependencies and services
134			services = set(services + dependencies)
135
136			# Modify existing AR specs with new form values of selected analyses.
137			self._update_specs(instance, specs)
138
139			# CREATE/MODIFY ANALYSES
140
141			for service in services:
142			keyword = service.getKeyword()
143
144			# Create the Analysis if it doesn't exist
145			if shasattr(instance, keyword):
146			analysis = instance._getOb(keyword)
147			else:
148			analysis = create_analysis(instance, service)
149			new_analyses.append(analysis)
150
151			# Set the price of the Analysis
152			self._update_price(analysis, service, prices)
153
154			# DELETE ANALYSES
155
156			# Service UIDs
157			service_uids = map(api.get_uid, services)
158
159			# Analyses IDs to delete
160			delete_ids = []
161
162			# Assigned Attachments
163			assigned_attachments = []
164
165			for analysis in analyses:
166			service_uid = analysis.getServiceUID()
167
168			# Skip if the Service is selected
169			if service_uid in service_uids:
170			continue
171
172			# Skip non-open Analyses
173			if analysis in non_open_analyses:
174			continue
175
176			# Remember assigned attachments
177			# https://github.com/senaite/senaite.core/issues/1025
178			assigned_attachments.extend(analysis.getAttachment())
179			analysis.setAttachment([])
180
181			# If it is assigned to a worksheet, unassign it before deletion.
182			worksheet = analysis.getWorksheet()
183			if worksheet:
184			worksheet.removeAnalysis(analysis)
185
186			# Unset the partition reference
187			# TODO Remove in >v1.3.0 - This is kept for backwards-compatibility
188			part = analysis.getSamplePartition()
189			if part:
190			# From this partition, remove the reference to the current
191			# analysis that is going to be removed to prevent inconsistent
192			# states (Sample Partitions referencing to Analyses that do not
193			# exist anymore
194			an_uid = api.get_uid(analysis)
195			part_ans = part.getAnalyses() or []
196			part_ans = filter(
197			lambda an: api.get_uid(an) != an_uid, part_ans)
			0 ignored issues – show introduced 2018-12-06 14:54 UTC by Report Bug Copy Issue Report The variable `an_uid` does not seem to be defined for all execution paths. Loading history...
198			part.setAnalyses(part_ans)
199			# Unset the Analysis-to-Partition reference
200			analysis.setSamplePartition(None)
201			delete_ids.append(analysis.getId())
202
203			if delete_ids:
204			# Note: subscriber might promote the AR
205			instance.manage_delObjects(ids=delete_ids)
206
207			# Remove orphaned attachments
208			for attachment in assigned_attachments:
209			# only delete attachments which are no further linked
210			if not attachment.getLinkedAnalyses():
211			logger.info(
212			"Deleting attachment: {}".format(attachment.getId()))
213			attachment_id = api.get_id(attachment)
214			api.get_parent(attachment).manage_delObjects(attachment_id)
215
216			return new_analyses
217
218			def _get_services(self, full_objects=False):
219			"""Fetch and return analysis service objects
220			"""
221			bsc = api.get_tool("bika_setup_catalog")
222			brains = bsc(portal_type="AnalysisService")
223			if full_objects:
224			return map(api.get_object, brains)
225			return brains
226
227			def _to_service(self, thing):
228			"""Convert to Analysis Service
229
230			:param thing: UID/Catalog Brain/Object/Something
231			:returns: Analysis Service object or None
232			"""
233
234			# Convert UIDs to objects
235			if api.is_uid(thing):
236			thing = api.get_object_by_uid(thing, None)
237
238			# Bail out if the thing is not a valid object
239			if not api.is_object(thing):
240			logger.warn("'{}' is not a valid object!".format(repr(thing)))
241			return None
242
243			# Ensure we have an object here and not a brain
244			obj = api.get_object(thing)
245
246			if IAnalysisService.providedBy(obj):
247			return obj
248
249			if IAnalysis.providedBy(obj):
250			return obj.getAnalysisService()
251
252			# An object, but neither an Analysis nor AnalysisService?
253			# This should never happen.
254			portal_type = api.get_portal_type(obj)
255			logger.error("ARAnalysesField doesn't accept objects from {} type. "
256			"The object will be dismissed.".format(portal_type))
257			return None
258
259			def _update_price(self, analysis, service, prices):
260			"""Update the Price of the Analysis
261
262			:param analysis: Analysis Object
263			:param service: Analysis Service Object
264			:param prices: Price mapping
265			"""
266			prices = prices or {}
267			price = prices.get(service.UID(), service.getPrice())
268			analysis.setPrice(price)
269
270			def _update_specs(self, instance, specs):
271			"""Update AR specifications
272
273			:param instance: Analysis Request
274			:param specs: List of Specification Records
275			"""
276
277			if specs is None:
278			return
279
280			# N.B. we copy the records here, otherwise the spec will be written to
281			# the attached specification of this AR
282			rr = {item["keyword"]: item.copy()
283			for item in instance.getResultsRange()}
284			for spec in specs:
285			keyword = spec.get("keyword")
286			if keyword in rr:
287			# overwrite the instance specification only, if the specific
288			# analysis spec has min/max values set
289			if all([spec.get("min"), spec.get("max")]):
290			rr[keyword].update(spec)
291			else:
292			rr[keyword] = spec
293			else:
294			rr[keyword] = spec
295			return instance.setResultsRange(rr.values())
296
297
298			registerField(ARAnalysesField,
299			title="Analyses",
300			description="Manages Analyses of ARs")
301

senaite / senaite.core

Push — master ( d8e2ec...90ae0b )

ARAnalysesField.set() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like