get_tmp_sample() - Code Metrics - Inspection of "Standalone samples and sample creation form" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#579)

by Pau

created 2018-01-17 17:28 UTC

get_tmp_sample() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	2
dl	0
loc	6
rs	9.4285
c	2
b	0
f	0

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

from Products.CMFPlone.utils import safe_unicode
from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile
from plone.memoize.volatile import cache
from zope.interface import implements
from zope.publisher.interfaces import IPublishTraverse

from bika.lims import api
from bika.lims import bikaMessageFactory as _
from bika.lims import logger
from bika.lims.browser.base_add_view import BaseAddView
from bika.lims.browser.base_add_view import BaseAjaxAddView
from bika.lims.browser.base_add_view import BaseManageAddView
from bika.lims.utils import cache_key
from bika.lims.utils.sample import create_sample


def get_tmp_sample(view):
    if not view.tmp_obj:
        logger.info("*** CREATING TEMPORARY SAMPLE ***")
        view.tmp_obj = view.context.restrictedTraverse(
            "portal_factory/Sample/sample_tmp")
    return view.tmp_obj


class SampleAddView(BaseAddView):
    """Sample Add view
    """
    template = ViewPageTemplateFile("templates/sample_add.pt")

    def __call__(self):
        BaseAddView.__call__(self)
        self.icon = self.portal_url + \
            "/++resource++bika.lims.images/sample_big.png"
        self.fieldvalues = self.generate_fieldvalues(self.obj_count)
        logger.info(
            "*** Prepared data for {} Samples ***".format(self.obj_count))
        return self.template()

    def get_obj(self):
        """Create a temporary Sample to fetch the fields from
        """
        self.tmp_obj = get_tmp_sample(self)
        return self.tmp_obj

    def generate_fieldvalues(self, count=1):
        """Returns a mapping of '<fieldname>-<count>' to the default value
        of the field or the field value of the source Sample
        """
        sample_context = self.get_obj()
        out = {}
        # the original schema fields of a Sample (including extended fields)
        fields = self.get_obj_fields()

        # generate fields for all requested Samples
        for samplenum in range(count):
            for field in fields:
                # get the default value of this field
                value = self.get_default_value(field, sample_context)
                # store the value on the new fieldname
                new_fieldname = self.get_fieldname(field, samplenum)
                out[new_fieldname] = value
        return out

    def get_object_by_uid(self, uid):
        """Get the object by UID
        """
        logger.debug("get_object_by_uid::UID={}".format(uid))
        obj = api.get_object_by_uid(uid, None)
        if obj is None:
            logger.warn("!! No object found for UID #{} !!")
        return obj

    def get_fields_with_visibility(self, visibility, mode="add"):
        """Return the Sample fields with the current visibility
        """
        sample = self.get_obj()
        mv = api.get_view("sample_add_manage", context=sample)
        mv.get_field_order()

        out = []
        for field in mv.get_fields_with_visibility(visibility, mode):
            # check custom field condition
            visible = self.is_field_visible(field)
            if visible is False and visibility != "hidden":
                continue
            out.append(field)
        return out


class SampleManageView(BaseManageAddView):
    """Sample Manage View
    """
    template = ViewPageTemplateFile("templates/sample_add_manage.pt")

    def __init__(self, context, request):
        BaseManageAddView.__init__(self, context, request)
        self.CONFIGURATION_STORAGE = "bika.lims.browser.sample.manage.add"

    def __call__(self):
        BaseManageAddView.__call__(self)
        return self.template()

    def get_obj(self):
        self.tmp_obj = get_tmp_sample(self)
        return self.tmp_obj


class ajaxSampleAdd(BaseAjaxAddView, SampleAddView):
    """Ajax helpers for the sample add form
    """
    implements(IPublishTraverse)

    def __init__(self, context, request):
        BaseAjaxAddView.__init__(self, context, request)

    @cache(cache_key)
    def get_client_info(self, obj):
        """Returns the client info of an object
        """
        info = self.get_base_info(obj)
        info.update({})

        # UID of the client
        uid = api.get_uid(obj)

        # Bika Setup folder
        bika_setup = api.get_bika_setup()

        # bika samplepoints
        bika_samplepoints = bika_setup.bika_samplepoints
        bika_samplepoints_uid = api.get_uid(bika_samplepoints)

        # catalog queries for UI field filtering
        filter_queries = {
            "samplepoint": {
                "getClientUID": [uid, bika_samplepoints_uid],
            },
        }
        info["filter_queries"] = filter_queries

        return info

    @cache(cache_key)

    def get_sampletype_info(self, obj):
        """Returns the info for a Sample Type
        """
        info = self.get_base_info(obj)

        # Bika Setup folder
        bika_setup = api.get_bika_setup()

        # bika samplepoints
        bika_samplepoints = bika_setup.bika_samplepoints
        bika_samplepoints_uid = api.get_uid(bika_samplepoints)

        # client
        client = self.get_client()
        client_uid = client and api.get_uid(client) or ""

        # sample matrix
        sample_matrix = obj.getSampleMatrix()
        sample_matrix_uid = sample_matrix and sample_matrix.UID() or ""
        sample_matrix_title = sample_matrix and sample_matrix.Title() or ""

        # container type
        container_type = obj.getContainerType()
        container_type_uid = container_type and container_type.UID() or ""
        container_type_title = container_type and container_type.Title() or ""

        # sample points
        sample_points = obj.getSamplePoints()
        sample_point_uids = map(lambda sp: sp.UID(), sample_points)
        sample_point_titles = map(lambda sp: sp.Title(), sample_points)

        info.update({
            "prefix": obj.getPrefix(),
            "minimum_volume": obj.getMinimumVolume(),
            "hazardous": obj.getHazardous(),
            "retention_period": obj.getRetentionPeriod(),
            "sample_matrix_uid": sample_matrix_uid,
            "sample_matrix_title": sample_matrix_title,
            "container_type_uid": container_type_uid,
            "container_type_title": container_type_title,
            "sample_point_uids": sample_point_uids,
            "sample_point_titles": sample_point_titles,
        })

        # catalog queries for UI field filtering
        filter_queries = {
            "samplepoint": {
                "getSampleTypeTitles": [obj.Title(), ''],
                "getClientUID": [client_uid, bika_samplepoints_uid],
                "sort_order": "descending",
            },
        }
        info["filter_queries"] = filter_queries

        return info

    def ajax_recalculate_records(self):
        """Recalculate all Sample records and dependencies

            - profiles
        """
        out = {}

        # The sorted records from the request
        records = self.get_records()

        for n, record in enumerate(records):

            # Mapping of client UID -> client object info
            client_metadata = {}
            # Mapping of sampletype UID -> sampletype object info
            sampletype_metadata = {}

            # Internal mappings of UID -> object of selected items in this
            # record
            _clients = self.get_objs_from_record(record, "Client_uid")
            _sampletypes = self.get_objs_from_record(record, "SampleType_uid")

            # CLIENTS
            for uid, obj in _clients.iteritems():
                # get the client metadata
                metadata = self.get_client_info(obj)
                # remember the sampletype metadata
                client_metadata[uid] = metadata

            # SAMPLETYPES
            for uid, obj in _sampletypes.iteritems():
                # get the sampletype metadata
                metadata = self.get_sampletype_info(obj)
                # remember the sampletype metadata
                sampletype_metadata[uid] = metadata


            # Each key `n` (1,2,3...) contains the form data for one
            # Sample Add column in the UI.
            # All relevant form data will be set according to this data.
            out[n] = {
                "client_metadata": client_metadata,
                "sampletype_metadata": sampletype_metadata,
            }

        return out

    def ajax_submit(self):
        """Submit & create the Samples
        """

        # Get Sample required fields (including extended fields)
        fields = self.get_obj_fields()

        # extract records from request
        records = self.get_records()

        fielderrors = {}
        errors = {"message": "", "fielderrors": {}}

        attachments = {}
        valid_records = []

        # Validate required fields
        for n, record in enumerate(records):

            # Process UID fields first and set their values to the linked field
            uid_fields = filter(lambda f: f.endswith("_uid"), record)
            for field in uid_fields:
                name = field.replace("_uid", "")
                value = record.get(field)
                if "," in value:
                    value = value.split(",")
                record[name] = value

            # Extract file uploads (fields ending with _file)
            # These files will be added later as attachments
            file_fields = filter(lambda f: f.endswith("_file"), record)
            attachments[n] = map(lambda f: record.pop(f), file_fields)


            # Required fields and their values
            required_keys = [field.getName() for field in fields if field.required]
            required_values = [record.get(key) for key in required_keys]
            required_fields = dict(zip(required_keys, required_values))

            # Client field is required but hidden in the sample Add form. We
            # remove
            # it therefore from the list of required fields to let empty
            # columns pass the required check below.
            if record.get("Client", False):
                required_fields.pop('Client', None)

            # None of the required fields are filled, skip this record
            if not any(required_fields.values()):
                continue

            # Missing required fields
            missing = [f for f in required_fields if not record.get(f, None)]

            # If there are required fields missing, flag an error
            for field in missing:
                fieldname = "{}-{}".format(field, n)
                msg = _("Field '{}' is required".format(field))
                fielderrors[fieldname] = msg

            # Process valid record
            valid_record = dict()
            for fieldname, fieldvalue in record.iteritems():
                # clean empty
                if fieldvalue in ['', None]:
                    continue
                valid_record[fieldname] = fieldvalue

            # append the valid record to the list of valid records
            valid_records.append(valid_record)

        # return immediately with an error response if some field checks failed
        if fielderrors:
            errors["fielderrors"] = fielderrors
            return {'errors': errors}

        # Process Form
        samples = []
        for n, record in enumerate(valid_records):
            client_uid = record.get("Client")
            client = self.get_object_by_uid(client_uid)

            if not client:
                raise RuntimeError("No client found")

            # Create the Sample
            try:
                sample = create_sample(client, self.request, record)
            except (KeyError, RuntimeError) as e:
                errors["message"] = e.message
                return {"errors": errors}
            samples.append(sample.Title())

        level = "info"
        if len(samples) == 0:
            message = _('No Sample could be created.')
            level = "error"
        elif len(samples) > 1:
            message = _('Samples ${samples} were successfully created.',
                        mapping={'Samples': safe_unicode(', '.join(samples))})
        else:
            message = _('Analysis request ${Sample} was successfully created.',
                        mapping={'Sample': safe_unicode(samples[0])})

        # Display a portal message
        self.context.plone_utils.addPortalMessage(message, level)

        bika_setup = api.get_bika_setup()
        auto_print = bika_setup.getAutoPrintStickers()

        # https://github.com/bikalabs/bika.lims/pull/2153
        new_samples = [a for a in samples if a[-1] == '1']

        if 'register' in auto_print and new_samples:
            return {
                'success': message,
                'stickers': new_samples,
                'stickertemplate': self.context.bika_setup.getAutoStickerTemplate()
            }
        else:
            return {'success': message}


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		from Products.CMFPlone.utils import safe_unicode
9		from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile
10		from plone.memoize.volatile import cache
11		from zope.interface import implements
12		from zope.publisher.interfaces import IPublishTraverse
13
14		from bika.lims import api
15		from bika.lims import bikaMessageFactory as _
16		from bika.lims import logger
17		from bika.lims.browser.base_add_view import BaseAddView
18		from bika.lims.browser.base_add_view import BaseAjaxAddView
19		from bika.lims.browser.base_add_view import BaseManageAddView
20		from bika.lims.utils import cache_key
21		from bika.lims.utils.sample import create_sample
22
23
24		def get_tmp_sample(view):
25		if not view.tmp_obj:
26		logger.info("* CREATING TEMPORARY SAMPLE *")
27		view.tmp_obj = view.context.restrictedTraverse(
28		"portal_factory/Sample/sample_tmp")
29		return view.tmp_obj
30
31
32		class SampleAddView(BaseAddView):
33		"""Sample Add view
34		"""
35		template = ViewPageTemplateFile("templates/sample_add.pt")
36
37		def __call__(self):
38		BaseAddView.__call__(self)
39		self.icon = self.portal_url + \
40		"/++resource++bika.lims.images/sample_big.png"
41		self.fieldvalues = self.generate_fieldvalues(self.obj_count)
42		logger.info(
43		"* Prepared data for {} Samples *".format(self.obj_count))
44		return self.template()
45
46		def get_obj(self):
47		"""Create a temporary Sample to fetch the fields from
48		"""
49		self.tmp_obj = get_tmp_sample(self)
50		return self.tmp_obj
51
52		def generate_fieldvalues(self, count=1):
53		"""Returns a mapping of '<fieldname>-<count>' to the default value
54		of the field or the field value of the source Sample
55		"""
56		sample_context = self.get_obj()
57		out = {}
58		# the original schema fields of a Sample (including extended fields)
59		fields = self.get_obj_fields()
60
61		# generate fields for all requested Samples
62		for samplenum in range(count):
63		for field in fields:
64		# get the default value of this field
65		value = self.get_default_value(field, sample_context)
66		# store the value on the new fieldname
67		new_fieldname = self.get_fieldname(field, samplenum)
68		out[new_fieldname] = value
69		return out
70
71		def get_object_by_uid(self, uid):
72		"""Get the object by UID
73		"""
74		logger.debug("get_object_by_uid::UID={}".format(uid))
75		obj = api.get_object_by_uid(uid, None)
76		if obj is None:
77		logger.warn("!! No object found for UID #{} !!")
78		return obj
79
80		def get_fields_with_visibility(self, visibility, mode="add"):
81		"""Return the Sample fields with the current visibility
82		"""
83		sample = self.get_obj()
84		mv = api.get_view("sample_add_manage", context=sample)
85		mv.get_field_order()
86
87		out = []
88		for field in mv.get_fields_with_visibility(visibility, mode):
89		# check custom field condition
90		visible = self.is_field_visible(field)
91		if visible is False and visibility != "hidden":
92		continue
93		out.append(field)
94		return out
95
96
97		class SampleManageView(BaseManageAddView):
98		"""Sample Manage View
99		"""
100		template = ViewPageTemplateFile("templates/sample_add_manage.pt")
101
102		def __init__(self, context, request):
103		BaseManageAddView.__init__(self, context, request)
104		self.CONFIGURATION_STORAGE = "bika.lims.browser.sample.manage.add"
105
106		def __call__(self):
107		BaseManageAddView.__call__(self)
108		return self.template()
109
110		def get_obj(self):
111		self.tmp_obj = get_tmp_sample(self)
112		return self.tmp_obj
113
114
115		class ajaxSampleAdd(BaseAjaxAddView, SampleAddView):
116		"""Ajax helpers for the sample add form
117		"""
118		implements(IPublishTraverse)
119
120		def __init__(self, context, request):
121		BaseAjaxAddView.__init__(self, context, request)
122
123		@cache(cache_key)
124		def get_client_info(self, obj):
125		"""Returns the client info of an object
126		"""
127		info = self.get_base_info(obj)
128		info.update({})
129
130		# UID of the client
131		uid = api.get_uid(obj)
132
133		# Bika Setup folder
134		bika_setup = api.get_bika_setup()
135
136		# bika samplepoints
137		bika_samplepoints = bika_setup.bika_samplepoints
138		bika_samplepoints_uid = api.get_uid(bika_samplepoints)
139
140		# catalog queries for UI field filtering
141		filter_queries = {
142		"samplepoint": {
143		"getClientUID": [uid, bika_samplepoints_uid],
144		},
145		}
146		info["filter_queries"] = filter_queries
147
148		return info
149
150	View Code Duplication	@cache(cache_key)
		0 ignored issues – show Duplication introduced 2018-01-17 17:30 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
151		def get_sampletype_info(self, obj):
152		"""Returns the info for a Sample Type
153		"""
154		info = self.get_base_info(obj)
155
156		# Bika Setup folder
157		bika_setup = api.get_bika_setup()
158
159		# bika samplepoints
160		bika_samplepoints = bika_setup.bika_samplepoints
161		bika_samplepoints_uid = api.get_uid(bika_samplepoints)
162
163		# client
164		client = self.get_client()
165		client_uid = client and api.get_uid(client) or ""
166
167		# sample matrix
168		sample_matrix = obj.getSampleMatrix()
169		sample_matrix_uid = sample_matrix and sample_matrix.UID() or ""
170		sample_matrix_title = sample_matrix and sample_matrix.Title() or ""
171
172		# container type
173		container_type = obj.getContainerType()
174		container_type_uid = container_type and container_type.UID() or ""
175		container_type_title = container_type and container_type.Title() or ""
176
177		# sample points
178		sample_points = obj.getSamplePoints()
179		sample_point_uids = map(lambda sp: sp.UID(), sample_points)
180		sample_point_titles = map(lambda sp: sp.Title(), sample_points)
181
182		info.update({
183		"prefix": obj.getPrefix(),
184		"minimum_volume": obj.getMinimumVolume(),
185		"hazardous": obj.getHazardous(),
186		"retention_period": obj.getRetentionPeriod(),
187		"sample_matrix_uid": sample_matrix_uid,
188		"sample_matrix_title": sample_matrix_title,
189		"container_type_uid": container_type_uid,
190		"container_type_title": container_type_title,
191		"sample_point_uids": sample_point_uids,
192		"sample_point_titles": sample_point_titles,
193		})
194
195		# catalog queries for UI field filtering
196		filter_queries = {
197		"samplepoint": {
198		"getSampleTypeTitles": [obj.Title(), ''],
199		"getClientUID": [client_uid, bika_samplepoints_uid],
200		"sort_order": "descending",
201		},
202		}
203		info["filter_queries"] = filter_queries
204
205		return info
206
207		def ajax_recalculate_records(self):
208		"""Recalculate all Sample records and dependencies
209
210		- profiles
211		"""
212		out = {}
213
214		# The sorted records from the request
215		records = self.get_records()
216
217		for n, record in enumerate(records):
218
219		# Mapping of client UID -> client object info
220		client_metadata = {}
221		# Mapping of sampletype UID -> sampletype object info
222		sampletype_metadata = {}
223
224		# Internal mappings of UID -> object of selected items in this
225		# record
226		_clients = self.get_objs_from_record(record, "Client_uid")
227		_sampletypes = self.get_objs_from_record(record, "SampleType_uid")
228
229		# CLIENTS
230		for uid, obj in _clients.iteritems():
231		# get the client metadata
232		metadata = self.get_client_info(obj)
233		# remember the sampletype metadata
234		client_metadata[uid] = metadata
235
236		# SAMPLETYPES
237		for uid, obj in _sampletypes.iteritems():
238		# get the sampletype metadata
239		metadata = self.get_sampletype_info(obj)
240		# remember the sampletype metadata
241		sampletype_metadata[uid] = metadata
242
243
244		# Each key `n` (1,2,3...) contains the form data for one
245		# Sample Add column in the UI.
246		# All relevant form data will be set according to this data.
247		out[n] = {
248		"client_metadata": client_metadata,
249		"sampletype_metadata": sampletype_metadata,
250		}
251
252		return out
253
254		def ajax_submit(self):
255		"""Submit & create the Samples
256		"""
257
258		# Get Sample required fields (including extended fields)
259		fields = self.get_obj_fields()
260
261		# extract records from request
262		records = self.get_records()
263
264		fielderrors = {}
265		errors = {"message": "", "fielderrors": {}}
266
267		attachments = {}
268		valid_records = []
269
270		# Validate required fields
271		for n, record in enumerate(records):
272
273		# Process UID fields first and set their values to the linked field
274		uid_fields = filter(lambda f: f.endswith("_uid"), record)
275		for field in uid_fields:
276		name = field.replace("_uid", "")
277		value = record.get(field)
278		if "," in value:
279		value = value.split(",")
280		record[name] = value
281
282		# Extract file uploads (fields ending with _file)
283		# These files will be added later as attachments
284		file_fields = filter(lambda f: f.endswith("_file"), record)
285		attachments[n] = map(lambda f: record.pop(f), file_fields)
		0 ignored issues – show Bug introduced 2018-01-17 17:30 UTC by Report Bug Copy Issue Report The loop variable `record` might not be defined here. Loading history...
286
287		# Required fields and their values
288		required_keys = [field.getName() for field in fields if field.required]
289		required_values = [record.get(key) for key in required_keys]
290		required_fields = dict(zip(required_keys, required_values))
291
292		# Client field is required but hidden in the sample Add form. We
293		# remove
294		# it therefore from the list of required fields to let empty
295		# columns pass the required check below.
296		if record.get("Client", False):
297		required_fields.pop('Client', None)
298
299		# None of the required fields are filled, skip this record
300		if not any(required_fields.values()):
301		continue
302
303		# Missing required fields
304		missing = [f for f in required_fields if not record.get(f, None)]
305
306		# If there are required fields missing, flag an error
307		for field in missing:
308		fieldname = "{}-{}".format(field, n)
309		msg = _("Field '{}' is required".format(field))
310		fielderrors[fieldname] = msg
311
312		# Process valid record
313		valid_record = dict()
314		for fieldname, fieldvalue in record.iteritems():
315		# clean empty
316		if fieldvalue in ['', None]:
317		continue
318		valid_record[fieldname] = fieldvalue
319
320		# append the valid record to the list of valid records
321		valid_records.append(valid_record)
322
323		# return immediately with an error response if some field checks failed
324		if fielderrors:
325		errors["fielderrors"] = fielderrors
326		return {'errors': errors}
327
328		# Process Form
329		samples = []
330		for n, record in enumerate(valid_records):
331		client_uid = record.get("Client")
332		client = self.get_object_by_uid(client_uid)
333
334		if not client:
335		raise RuntimeError("No client found")
336
337		# Create the Sample
338		try:
339		sample = create_sample(client, self.request, record)
340		except (KeyError, RuntimeError) as e:
341		errors["message"] = e.message
342		return {"errors": errors}
343		samples.append(sample.Title())
344
345		level = "info"
346		if len(samples) == 0:
347		message = _('No Sample could be created.')
348		level = "error"
349		elif len(samples) > 1:
350		message = _('Samples ${samples} were successfully created.',
351		mapping={'Samples': safe_unicode(', '.join(samples))})
352		else:
353		message = _('Analysis request ${Sample} was successfully created.',
354		mapping={'Sample': safe_unicode(samples[0])})
355
356		# Display a portal message
357		self.context.plone_utils.addPortalMessage(message, level)
358
359		bika_setup = api.get_bika_setup()
360		auto_print = bika_setup.getAutoPrintStickers()
361
362		# https://github.com/bikalabs/bika.lims/pull/2153
363		new_samples = [a for a in samples if a[-1] == '1']
364
365		if 'register' in auto_print and new_samples:
366		return {
367		'success': message,
368		'stickers': new_samples,
369		'stickertemplate': self.context.bika_setup.getAutoStickerTemplate()
370		}
371		else:
372		return {'success': message}
373

senaite / senaite.core

Pull Request — master (#579)

get_tmp_sample() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like