bika.lims.utils.analysisrequest.get_services_uids() - Code Metrics - Inspection of "Sample cleanup (#1180)" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Branch — master (9edffc)

by Jordi

created 2019-01-04 23:53 UTC

get_services_uids() C

↳ Parent: bika.lims.utils.analysisrequest

Complexity

Conditions

Size

Total Lines	50
Code Lines	22

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	22
dl	0
loc	50
rs	5.4
c	0
b	0
f	0
cc	11
nop	3

How to fix Complexity

# -*- 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 os
import tempfile
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

from Products.CMFCore.utils import getToolByName
from Products.CMFPlone.utils import _createObjectByType
from Products.CMFPlone.utils import safe_unicode
from bika.lims import api
from bika.lims import bikaMessageFactory as _
from bika.lims import logger
from bika.lims.idserver import renameAfterCreation
from bika.lims.interfaces import IAnalysisService, IRoutineAnalysis, \
    IAnalysisRequest
from bika.lims.utils import attachPdf
from bika.lims.utils import changeWorkflowState
from bika.lims.utils import createPdf
from bika.lims.utils import encode_header
from bika.lims.utils import tmpID, copy_field_values
from bika.lims.utils import to_utf8
from bika.lims.workflow import doActionFor, ActionHandlerPool, \
    push_reindex_to_actions_pool
from email.Utils import formataddr


def create_analysisrequest(client, request, values, analyses=None,
                           partitions=None, specifications=None, prices=None):
    """This is meant for general use and should do everything necessary to
    create and initialise an AR and any other required auxilliary objects
    (Sample, SamplePartition, Analysis...)
    :param client:
        The container (Client) in which the ARs will be created.
    :param request:
        The current Request object.
    :param values:
        a dict, where keys are AR|Sample schema field names.
    :param analyses:
        Analysis services list.  If specified, augments the values in
        values['Analyses']. May consist of service objects, UIDs, or Keywords.
    :param partitions:
        A list of dictionaries, if specific partitions are required.  If not
        specified, AR's sample is created with a single partition.
    :param specifications:
        These values augment those found in values['Specifications']
    :param prices:
        Allow different prices to be set for analyses.  If not set, prices
        are read from the associated analysis service.
    """
    # Don't pollute the dict param passed in
    values = dict(values.items())

    # Create new sample or locate the existing for secondary AR
    secondary = False
    # TODO Sample Cleanup - Manage secondary ARs properly

    # Create the Analysis Request
    ar = _createObjectByType('AnalysisRequest', client, tmpID())
    ar.processForm(REQUEST=request, values=values)

    # Resolve the services uids and set the analyses for this Analysis Request
    service_uids = get_services_uids(context=client, values=values,
                                     analyses_serv=analyses)
    ar.setAnalyses(service_uids, prices=prices, specs=specifications)

    # TODO Sample Cleanup - Manage secondary ARs properly
    if secondary:
        # Secondary AR does not longer comes from a Sample, rather from an AR.
        # If the Primary AR has been received, then force the transition of the
        # secondary to received and set the description/comment in the
        # transition accordingly so it will be displayed later in the log tab
        logger.warn("Sync transition for secondary AR is still missing")

    # Needs to be rejected from the very beginning?
    reject_field = values.get("RejectionReasons", None)
    if reject_field and reject_field.get("checkbox", False):
        doActionFor(ar, "reject")
        return ar

    # Try first with no sampling transition, cause it is the most common config
    success, message = doActionFor(ar, "no_sampling_workflow")
    if not success:
        doActionFor(ar, "to_be_sampled")
    return ar


def get_services_uids(context=None, analyses_serv=None, values=None):
    """
    This function returns a list of UIDs from analyses services from its
    parameters.
    :param analyses_serv: A list (or one object) of service-related info items.
        see _resolve_items_to_service_uids() docstring.
    :type analyses_serv: list
    :param values: a dict, where keys are AR|Sample schema field names.
    :type values: dict
    :returns: a list of analyses services UIDs
    """
    if not analyses_serv:
        analyses_serv = []
    if not values:
        values = {}

    if not context or (not analyses_serv and not values):
        raise RuntimeError(
            "get_services_uids: Missing or wrong parameters.")

    # Merge analyses from analyses_serv and values into one list
    analyses_services = analyses_serv + (values.get("Analyses", None) or [])

    # It is possible to create analysis requests
    # by JSON petitions and services, profiles or types aren't allways send.
    # Sometimes we can get analyses and profiles that doesn't match and we
    # should act in consequence.
    # Getting the analyses profiles
    analyses_profiles = values.get('Profiles', [])
    if not isinstance(analyses_profiles, (list, tuple)):
        # Plone converts the incoming form value to a list, if there are
        # multiple values; but if not, it will send a string (a single UID).
        analyses_profiles = [analyses_profiles]

    if not analyses_services and not analyses_profiles:
        raise RuntimeError(
                "create_analysisrequest: no analyses services or analysis"
                " profile provided")

    # Add analysis services UIDs from profiles to analyses_services variable.
    if analyses_profiles:
        uid_catalog = getToolByName(context, 'uid_catalog')
        for brain in uid_catalog(UID=analyses_profiles):
            profile = api.get_object(brain)
            # Only services UIDs
            services_uids = profile.getRawService()
            # _resolve_items_to_service_uids() will remove duplicates
            analyses_services += services_uids

    return _resolve_items_to_service_uids(analyses_services)


def _resolve_items_to_service_uids(items):
    """ Returns a list of service uids without duplicates based on the items
    :param items:
        A list (or one object) of service-related info items. The list can be
        heterogeneous and each item can be:
        - Analysis Service instance
        - Analysis instance
        - Analysis Service title
        - Analysis Service UID
        - Analysis Service Keyword
        If an item that doesn't match any of the criterias above is found, the
        function will raise a RuntimeError
    """
    def resolve_to_uid(item):
        if api.is_uid(item):
            return item
        elif IAnalysisService.providedBy(item):
            return item.UID()
        elif IRoutineAnalysis.providedBy(item):
            return item.getServiceUID()

        bsc = api.get_tool("bika_setup_catalog")
        brains = bsc(portal_type='AnalysisService', getKeyword=item)
        if brains:
            return brains[0].UID
        brains = bsc(portal_type='AnalysisService', title=item)
        if brains:
            return brains[0].UID
        raise RuntimeError(
            str(item) + " should be the UID, title, keyword "
                        " or title of an AnalysisService.")

    # Maybe only a single item was passed
    if type(items) not in (list, tuple):
        items = [items, ]
    service_uids = map(resolve_to_uid, list(set(items)))
    return list(set(service_uids))


def notify_rejection(analysisrequest):
    """
    Notifies via email that a given Analysis Request has been rejected. The
    notification is sent to the Client contacts assigned to the Analysis
    Request.

    :param analysisrequest: Analysis Request to which the notification refers
    :returns: true if success
    """

    # We do this imports here to avoid circular dependencies until we deal
    # better with this notify_rejection thing.
    from bika.lims.browser.analysisrequest.reject import \
        AnalysisRequestRejectPdfView, AnalysisRequestRejectEmailView

    arid = analysisrequest.getId()

    # This is the template to render for the pdf that will be either attached
    # to the email and attached the the Analysis Request for further access
    tpl = AnalysisRequestRejectPdfView(analysisrequest, analysisrequest.REQUEST)
    html = tpl.template()
    html = safe_unicode(html).encode('utf-8')
    filename = '%s-rejected' % arid
    pdf_fn = tempfile.mktemp(suffix=".pdf")
    pdf = createPdf(htmlreport=html, outfile=pdf_fn)
    if pdf:
        # Attach the pdf to the Analysis Request
        attid = analysisrequest.aq_parent.generateUniqueId('Attachment')
        att = _createObjectByType(
            "Attachment", analysisrequest.aq_parent, attid)
        att.setAttachmentFile(open(pdf_fn))
        # Awkward workaround to rename the file
        attf = att.getAttachmentFile()
        attf.filename = '%s.pdf' % filename
        att.setAttachmentFile(attf)
        att.unmarkCreationFlag()
        renameAfterCreation(att)
        atts = analysisrequest.getAttachment() + [att] if \
            analysisrequest.getAttachment() else [att]
        atts = [a.UID() for a in atts]
        analysisrequest.setAttachment(atts)
        os.remove(pdf_fn)

    # This is the message for the email's body
    tpl = AnalysisRequestRejectEmailView(
        analysisrequest, analysisrequest.REQUEST)
    html = tpl.template()
    html = safe_unicode(html).encode('utf-8')

    # compose and send email.
    mailto = []
    lab = analysisrequest.bika_setup.laboratory
    mailfrom = formataddr((encode_header(lab.getName()), lab.getEmailAddress()))
    mailsubject = _('%s has been rejected') % arid
    contacts = [analysisrequest.getContact()] + analysisrequest.getCCContact()
    for contact in contacts:
        name = to_utf8(contact.getFullname())
        email = to_utf8(contact.getEmailAddress())
        if email:
            mailto.append(formataddr((encode_header(name), email)))
    if not mailto:
        return False
    mime_msg = MIMEMultipart('related')
    mime_msg['Subject'] = mailsubject
    mime_msg['From'] = mailfrom
    mime_msg['To'] = ','.join(mailto)
    mime_msg.preamble = 'This is a multi-part MIME message.'
    msg_txt = MIMEText(html, _subtype='html')
    mime_msg.attach(msg_txt)
    if pdf:
        attachPdf(mime_msg, pdf, filename)

    try:
        host = getToolByName(analysisrequest, 'MailHost')
        host.send(mime_msg.as_string(), immediate=True)
    except:
        logger.warning(
            "Email with subject %s was not sent (SMTP connection error)" % mailsubject)

    return True


def create_retest(ar):
    """Creates a retest (Analysis Request) from an invalidated Analysis Request
    :param ar: The invalidated Analysis Request
    :type ar: IAnalysisRequest
    :rtype: IAnalysisRequest
    """
    if not ar:
        raise ValueError("Source Analysis Request cannot be None")

    if not IAnalysisRequest.providedBy(ar):
        raise ValueError("Type not supported: {}".format(repr(type(ar))))

    if ar.getRetest():
        # Do not allow the creation of another retest!
        raise ValueError("Retest already set")

    if not ar.isInvalid():
        # Analysis Request must be in 'invalid' state
        raise ValueError("Cannot do a retest from an invalid Analysis Request"
                         .format(repr(ar)))

    # 0. Open the actions pool
    actions_pool = ActionHandlerPool.get_instance()
    actions_pool.queue_pool()

    # 1. Create the Retest (Analysis Request)
    ignore = ['Analyses', 'DatePublished', 'Invalidated', 'Sample']
    retest = _createObjectByType("AnalysisRequest", ar.aq_parent, tmpID())
    copy_field_values(ar, retest, ignore_fieldnames=ignore)
    renameAfterCreation(retest)

    # 2. Copy the analyses from the source
    intermediate_states = ['retracted', 'reflexed']
    for an in ar.getAnalyses(full_objects=True):
        if (api.get_workflow_status_of(an) in intermediate_states):
            # Exclude intermediate analyses
            continue

        nan = _createObjectByType("Analysis", retest, an.getKeyword())

        # Make a copy
        ignore_fieldnames = ['DataAnalysisPublished']
        copy_field_values(an, nan, ignore_fieldnames=ignore_fieldnames)
        nan.unmarkCreationFlag()
        push_reindex_to_actions_pool(nan)

    # 3. Assign the source to retest
    retest.setInvalidated(ar)

    # 4. Transition the retest to "sample_received"!
    changeWorkflowState(retest, 'bika_ar_workflow', 'sample_received')

    # 5. Reindex and other stuff
    push_reindex_to_actions_pool(retest)
    push_reindex_to_actions_pool(retest.aq_parent)

    # 6. Resume the actions pool
    actions_pool.resume()
    return retest


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 os
9			import tempfile
10			from email.mime.multipart import MIMEMultipart
11			from email.mime.text import MIMEText
12
13			from Products.CMFCore.utils import getToolByName
14			from Products.CMFPlone.utils import _createObjectByType
15			from Products.CMFPlone.utils import safe_unicode
16			from bika.lims import api
17			from bika.lims import bikaMessageFactory as _
18			from bika.lims import logger
19			from bika.lims.idserver import renameAfterCreation
20			from bika.lims.interfaces import IAnalysisService, IRoutineAnalysis, \
21			IAnalysisRequest
22			from bika.lims.utils import attachPdf
23			from bika.lims.utils import changeWorkflowState
24			from bika.lims.utils import createPdf
25			from bika.lims.utils import encode_header
26			from bika.lims.utils import tmpID, copy_field_values
27			from bika.lims.utils import to_utf8
28			from bika.lims.workflow import doActionFor, ActionHandlerPool, \
29			push_reindex_to_actions_pool
30			from email.Utils import formataddr
31
32
33			def create_analysisrequest(client, request, values, analyses=None,
34			partitions=None, specifications=None, prices=None):
35			"""This is meant for general use and should do everything necessary to
36			create and initialise an AR and any other required auxilliary objects
37			(Sample, SamplePartition, Analysis...)
38			:param client:
39			The container (Client) in which the ARs will be created.
40			:param request:
41			The current Request object.
42			:param values:
43			a dict, where keys are AR\|Sample schema field names.
44			:param analyses:
45			Analysis services list. If specified, augments the values in
46			values['Analyses']. May consist of service objects, UIDs, or Keywords.
47			:param partitions:
48			A list of dictionaries, if specific partitions are required. If not
49			specified, AR's sample is created with a single partition.
50			:param specifications:
51			These values augment those found in values['Specifications']
52			:param prices:
53			Allow different prices to be set for analyses. If not set, prices
54			are read from the associated analysis service.
55			"""
56			# Don't pollute the dict param passed in
57			values = dict(values.items())
58
59			# Create new sample or locate the existing for secondary AR
60			secondary = False
61			# TODO Sample Cleanup - Manage secondary ARs properly
62
63			# Create the Analysis Request
64			ar = _createObjectByType('AnalysisRequest', client, tmpID())
65			ar.processForm(REQUEST=request, values=values)
66
67			# Resolve the services uids and set the analyses for this Analysis Request
68			service_uids = get_services_uids(context=client, values=values,
69			analyses_serv=analyses)
70			ar.setAnalyses(service_uids, prices=prices, specs=specifications)
71
72			# TODO Sample Cleanup - Manage secondary ARs properly
73			if secondary:
74			# Secondary AR does not longer comes from a Sample, rather from an AR.
75			# If the Primary AR has been received, then force the transition of the
76			# secondary to received and set the description/comment in the
77			# transition accordingly so it will be displayed later in the log tab
78			logger.warn("Sync transition for secondary AR is still missing")
79
80			# Needs to be rejected from the very beginning?
81			reject_field = values.get("RejectionReasons", None)
82			if reject_field and reject_field.get("checkbox", False):
83			doActionFor(ar, "reject")
84			return ar
85
86			# Try first with no sampling transition, cause it is the most common config
87			success, message = doActionFor(ar, "no_sampling_workflow")
88			if not success:
89			doActionFor(ar, "to_be_sampled")
90			return ar
91
92
93			def get_services_uids(context=None, analyses_serv=None, values=None):
94			"""
95			This function returns a list of UIDs from analyses services from its
96			parameters.
97			:param analyses_serv: A list (or one object) of service-related info items.
98			see _resolve_items_to_service_uids() docstring.
99			:type analyses_serv: list
100			:param values: a dict, where keys are AR\|Sample schema field names.
101			:type values: dict
102			:returns: a list of analyses services UIDs
103			"""
104			if not analyses_serv:
105			analyses_serv = []
106			if not values:
107			values = {}
108
109			if not context or (not analyses_serv and not values):
110			raise RuntimeError(
111			"get_services_uids: Missing or wrong parameters.")
112
113			# Merge analyses from analyses_serv and values into one list
114			analyses_services = analyses_serv + (values.get("Analyses", None) or [])
115
116			# It is possible to create analysis requests
117			# by JSON petitions and services, profiles or types aren't allways send.
118			# Sometimes we can get analyses and profiles that doesn't match and we
119			# should act in consequence.
120			# Getting the analyses profiles
121			analyses_profiles = values.get('Profiles', [])
122			if not isinstance(analyses_profiles, (list, tuple)):
123			# Plone converts the incoming form value to a list, if there are
124			# multiple values; but if not, it will send a string (a single UID).
125			analyses_profiles = [analyses_profiles]
126
127			if not analyses_services and not analyses_profiles:
128			raise RuntimeError(
129			"create_analysisrequest: no analyses services or analysis"
130			" profile provided")
131
132			# Add analysis services UIDs from profiles to analyses_services variable.
133			if analyses_profiles:
134			uid_catalog = getToolByName(context, 'uid_catalog')
135			for brain in uid_catalog(UID=analyses_profiles):
136			profile = api.get_object(brain)
137			# Only services UIDs
138			services_uids = profile.getRawService()
139			# _resolve_items_to_service_uids() will remove duplicates
140			analyses_services += services_uids
141
142			return _resolve_items_to_service_uids(analyses_services)
143
144
145			def _resolve_items_to_service_uids(items):
146			""" Returns a list of service uids without duplicates based on the items
147			:param items:
148			A list (or one object) of service-related info items. The list can be
149			heterogeneous and each item can be:
150			- Analysis Service instance
151			- Analysis instance
152			- Analysis Service title
153			- Analysis Service UID
154			- Analysis Service Keyword
155			If an item that doesn't match any of the criterias above is found, the
156			function will raise a RuntimeError
157			"""
158			def resolve_to_uid(item):
159			if api.is_uid(item):
160			return item
161			elif IAnalysisService.providedBy(item):
162			return item.UID()
163			elif IRoutineAnalysis.providedBy(item):
164			return item.getServiceUID()
165
166			bsc = api.get_tool("bika_setup_catalog")
167			brains = bsc(portal_type='AnalysisService', getKeyword=item)
168			if brains:
169			return brains[0].UID
170			brains = bsc(portal_type='AnalysisService', title=item)
171			if brains:
172			return brains[0].UID
173			raise RuntimeError(
174			str(item) + " should be the UID, title, keyword "
175			" or title of an AnalysisService.")
176
177			# Maybe only a single item was passed
178			if type(items) not in (list, tuple):
179			items = [items, ]
180			service_uids = map(resolve_to_uid, list(set(items)))
181			return list(set(service_uids))
182
183
184			def notify_rejection(analysisrequest):
185			"""
186			Notifies via email that a given Analysis Request has been rejected. The
187			notification is sent to the Client contacts assigned to the Analysis
188			Request.
189
190			:param analysisrequest: Analysis Request to which the notification refers
191			:returns: true if success
192			"""
193
194			# We do this imports here to avoid circular dependencies until we deal
195			# better with this notify_rejection thing.
196			from bika.lims.browser.analysisrequest.reject import \
197			AnalysisRequestRejectPdfView, AnalysisRequestRejectEmailView
198
199			arid = analysisrequest.getId()
200
201			# This is the template to render for the pdf that will be either attached
202			# to the email and attached the the Analysis Request for further access
203			tpl = AnalysisRequestRejectPdfView(analysisrequest, analysisrequest.REQUEST)
204			html = tpl.template()
205			html = safe_unicode(html).encode('utf-8')
206			filename = '%s-rejected' % arid
207			pdf_fn = tempfile.mktemp(suffix=".pdf")
208			pdf = createPdf(htmlreport=html, outfile=pdf_fn)
209			if pdf:
210			# Attach the pdf to the Analysis Request
211			attid = analysisrequest.aq_parent.generateUniqueId('Attachment')
212			att = _createObjectByType(
213			"Attachment", analysisrequest.aq_parent, attid)
214			att.setAttachmentFile(open(pdf_fn))
215			# Awkward workaround to rename the file
216			attf = att.getAttachmentFile()
217			attf.filename = '%s.pdf' % filename
218			att.setAttachmentFile(attf)
219			att.unmarkCreationFlag()
220			renameAfterCreation(att)
221			atts = analysisrequest.getAttachment() + [att] if \
222			analysisrequest.getAttachment() else [att]
223			atts = [a.UID() for a in atts]
224			analysisrequest.setAttachment(atts)
225			os.remove(pdf_fn)
226
227			# This is the message for the email's body
228			tpl = AnalysisRequestRejectEmailView(
229			analysisrequest, analysisrequest.REQUEST)
230			html = tpl.template()
231			html = safe_unicode(html).encode('utf-8')
232
233			# compose and send email.
234			mailto = []
235			lab = analysisrequest.bika_setup.laboratory
236			mailfrom = formataddr((encode_header(lab.getName()), lab.getEmailAddress()))
237			mailsubject = _('%s has been rejected') % arid
238			contacts = [analysisrequest.getContact()] + analysisrequest.getCCContact()
239			for contact in contacts:
240			name = to_utf8(contact.getFullname())
241			email = to_utf8(contact.getEmailAddress())
242			if email:
243			mailto.append(formataddr((encode_header(name), email)))
244			if not mailto:
245			return False
246			mime_msg = MIMEMultipart('related')
247			mime_msg['Subject'] = mailsubject
248			mime_msg['From'] = mailfrom
249			mime_msg['To'] = ','.join(mailto)
250			mime_msg.preamble = 'This is a multi-part MIME message.'
251			msg_txt = MIMEText(html, _subtype='html')
252			mime_msg.attach(msg_txt)
253			if pdf:
254			attachPdf(mime_msg, pdf, filename)
255
256			try:
257			host = getToolByName(analysisrequest, 'MailHost')
258			host.send(mime_msg.as_string(), immediate=True)
259			except:
260			logger.warning(
261			"Email with subject %s was not sent (SMTP connection error)" % mailsubject)
262
263			return True
264
265
266			def create_retest(ar):
267			"""Creates a retest (Analysis Request) from an invalidated Analysis Request
268			:param ar: The invalidated Analysis Request
269			:type ar: IAnalysisRequest
270			:rtype: IAnalysisRequest
271			"""
272			if not ar:
273			raise ValueError("Source Analysis Request cannot be None")
274
275			if not IAnalysisRequest.providedBy(ar):
276			raise ValueError("Type not supported: {}".format(repr(type(ar))))
277
278			if ar.getRetest():
279			# Do not allow the creation of another retest!
280			raise ValueError("Retest already set")
281
282			if not ar.isInvalid():
283			# Analysis Request must be in 'invalid' state
284			raise ValueError("Cannot do a retest from an invalid Analysis Request"
285			.format(repr(ar)))
286
287			# 0. Open the actions pool
288			actions_pool = ActionHandlerPool.get_instance()
289			actions_pool.queue_pool()
290
291			# 1. Create the Retest (Analysis Request)
292			ignore = ['Analyses', 'DatePublished', 'Invalidated', 'Sample']
293			retest = _createObjectByType("AnalysisRequest", ar.aq_parent, tmpID())
294			copy_field_values(ar, retest, ignore_fieldnames=ignore)
295			renameAfterCreation(retest)
296
297			# 2. Copy the analyses from the source
298			intermediate_states = ['retracted', 'reflexed']
299			for an in ar.getAnalyses(full_objects=True):
300			if (api.get_workflow_status_of(an) in intermediate_states):
301			# Exclude intermediate analyses
302			continue
303
304			nan = _createObjectByType("Analysis", retest, an.getKeyword())
305
306			# Make a copy
307			ignore_fieldnames = ['DataAnalysisPublished']
308			copy_field_values(an, nan, ignore_fieldnames=ignore_fieldnames)
309			nan.unmarkCreationFlag()
310			push_reindex_to_actions_pool(nan)
311
312			# 3. Assign the source to retest
313			retest.setInvalidated(ar)
314
315			# 4. Transition the retest to "sample_received"!
316			changeWorkflowState(retest, 'bika_ar_workflow', 'sample_received')
317
318			# 5. Reindex and other stuff
319			push_reindex_to_actions_pool(retest)
320			push_reindex_to_actions_pool(retest.aq_parent)
321
322			# 6. Resume the actions pool
323			actions_pool.resume()
324			return retest
325

senaite / senaite.core

Branch — master (9edffc)

get_services_uids() C

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like