bika.lims.jsonapi.set_fields_from_request() - Code Metrics - Inspection of "Fix jsonapi.read omits `include_methods` when a si..." - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 2280ab...e4f397 )

by Ramon

created 2019-12-23 06:58 UTC

bika.lims.jsonapi.set_fields_from_request() F

↳ Parent: bika.lims.jsonapi

Complexity

Conditions

Size

Total Lines	54
Code Lines	42

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	42
dl	0
loc	54
rs	2.4
c	0
b	0
f	0
cc	16
nop	2

How to fix Long Method Complexity

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

from Products.Archetypes.config import TOOL_NAME
from Products.CMFCore.utils import getToolByName
from bika.lims.utils import to_utf8
from bika.lims import logger

import json
import Missing
import six
import sys, traceback


def handle_errors(f):
    """ simple JSON error handler
    """
    import traceback
    from plone.jsonapi.core.helpers import error

    def decorator(*args, **kwargs):
        try:
            return f(*args, **kwargs)
        except Exception:
            var = traceback.format_exc()
            return error(var)

    return decorator


def get_include_fields(request):
    """Retrieve include_fields values from the request
    """
    include_fields = []
    rif = request.get("include_fields", "")
    if "include_fields" in request:
        include_fields = [x.strip()
                          for x in rif.split(",")
                          if x.strip()]
    if "include_fields[]" in request:
        include_fields = request['include_fields[]']
    return include_fields


def load_brain_metadata(proxy, include_fields):
    """Load values from the catalog metadata into a list of dictionaries
    """
    ret = {}
    for index in proxy.indexes():
        if index not in proxy:
            continue
        if include_fields and index not in include_fields:
            continue
        val = getattr(proxy, index)
        if val != Missing.Value:
            try:
                json.dumps(val)
            except:
                continue
            ret[index] = val
    return ret


def load_field_values(instance, include_fields):
    """Load values from an AT object schema fields into a list of dictionaries
    """
    ret = {}
    schema = instance.Schema()
    val = None
    for field in schema.fields():
        fieldname = field.getName()
        if include_fields and fieldname not in include_fields:
            continue
        try:
            val = field.get(instance)
        except AttributeError:
            # If this error is raised, make a look to the add-on content
            # expressions used to obtain their data.
            print "AttributeError:", sys.exc_info()[1]
            print "Unreachable object. Maybe the object comes from an Add-on"
            print traceback.format_exc()

        if val:
            field_type = field.type
            # If it a proxy field, we should know to the type of the proxied
            # field
            if field_type == 'proxy':
                actual_field = field.get_proxy(instance)
                field_type = actual_field.type
            if field_type == "blob" or field_type == 'file':
                continue
            # I put the UID of all references here in *_uid.
            if field_type == 'reference':
                if type(val) in (list, tuple):
                    ret[fieldname + "_uid"] = [v.UID() for v in val]
                    val = [to_utf8(v.Title()) for v in val]
                else:
                    ret[fieldname + "_uid"] = val.UID()
                    val = to_utf8(val.Title())
            elif field_type == 'boolean':
                val = True if val else False
            elif field_type == 'text':
                val = to_utf8(val)

        try:
            json.dumps(val)
        except:
            val = str(val)
        ret[fieldname] = val
    return ret


def get_include_methods(request):
    """Retrieve include_methods values from the request
    """
    include_methods = request.get("include_methods[]")
    if not include_methods:
        include_methods = request.get("include_methods", [])

    if isinstance(include_methods, six.string_types):
        include_methods = include_methods.split(",")
        include_methods = map(lambda me: me.strip(), include_methods)

    return filter(None, include_methods)


def load_method_values(instance, include_methods):
    ret = {}
    for method in include_methods:
        if hasattr(instance, method):
            val = getattr(instance, method)()
            ret[method] = val
    return ret


def resolve_request_lookup(context, request, fieldname):
    if fieldname not in request:
        return []
    brains = []
    at = getToolByName(context, TOOL_NAME, None)
    entries = request[fieldname] if type(request[fieldname]) in (list, tuple) \
        else [request[fieldname], ]
    for entry in entries:
        contentFilter = {}
        for value in entry.split("|"):
            if ":" in value:
                index, value = value.split(":", 1)
            else:
                index, value = 'id', value
            if index in contentFilter:
                v = contentFilter[index]
                v = v if type(v) in (list, tuple) else [v, ]
                v.append(value)
                contentFilter[index] = v
            else:
                contentFilter[index] = value
        # search all possible catalogs
        if 'portal_type' in contentFilter:
            catalogs = at.getCatalogsByType(contentFilter['portal_type'])
        elif 'uid' in contentFilter:
            # If a uid is found, the object could be searched for its own uid
            uc = getToolByName(context, 'uid_catalog')
            return uc(UID=contentFilter['uid'])

        else:
            catalogs = [getToolByName(context, 'portal_catalog'), ]
        for catalog in catalogs:
            _brains = catalog(contentFilter)
            if _brains:
                brains.extend(_brains)
                break
    return brains


def set_fields_from_request(obj, request):
    """Search request for keys that match field names in obj,
    and call field mutator with request value.

    The list of fields for which schema mutators were found
    is returned.
    """
    schema = obj.Schema()
    # fields contains all schema-valid field values from the request.
    fields = {}
    for fieldname, value in request.items():
        if fieldname not in schema:
            continue
        field = schema[fieldname]
        widget = field.getWidgetName()
        if widget in ["ReferenceWidget"]:
            brains = []
            if value:
                brains = resolve_request_lookup(obj, request, fieldname)
                if not brains:
                    logger.warning(
                        "JSONAPI: Can't resolve reference: {} {}"
                        .format(fieldname, value))
                    return []
            if schema[fieldname].multiValued:
                value = [b.UID for b in brains] if brains else []
            else:
                value = brains[0].UID if brains else None
        fields[fieldname] = value
    # Write fields.
    for fieldname, value in fields.items():
        field = schema[fieldname]
        fieldtype = field.getType()
        if fieldtype == 'Products.Archetypes.Field.BooleanField':
            if value.lower() in ('0', 'false', 'no') or not value:
                value = False
            else:
                value = True
        elif fieldtype in ['Products.ATExtensions.field.records.RecordsField',
                           'Products.ATExtensions.field.records.RecordField']:
            try:
                value = eval(value)
            except:
                logger.warning(
                    "JSONAPI: " + fieldname + ": Invalid "
                    "JSON/Python variable")
                return []
        mutator = field.getMutator(obj)
        if mutator:
            mutator(value)
        else:
            field.set(obj, value)
    obj.reindexObject()
    return fields.keys()


1			# -- coding: utf-8 --
2			#
3			# This file is part of SENAITE.CORE.
4			#
5			# SENAITE.CORE is free software: you can redistribute it and/or modify it under
6			# the terms of the GNU General Public License as published by the Free Software
7			# Foundation, version 2.
8			#
9			# This program is distributed in the hope that it will be useful, but WITHOUT
10			# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
11			# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
12			# details.
13			#
14			# You should have received a copy of the GNU General Public License along with
15			# this program; if not, write to the Free Software Foundation, Inc., 51
16			# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17			#
18			# Copyright 2018-2019 by it's authors.
19			# Some rights reserved, see README and LICENSE.
20
21			from Products.Archetypes.config import TOOL_NAME
22			from Products.CMFCore.utils import getToolByName
23			from bika.lims.utils import to_utf8
24			from bika.lims import logger
25
26			import json
27			import Missing
28			import six
29			import sys, traceback
30
31
32			def handle_errors(f):
33			""" simple JSON error handler
34			"""
35			import traceback
36			from plone.jsonapi.core.helpers import error
37
38			def decorator(args, *kwargs):
39			try:
40			return f(args, *kwargs)
41			except Exception:
42			var = traceback.format_exc()
43			return error(var)
44
45			return decorator
46
47
48			def get_include_fields(request):
49			"""Retrieve include_fields values from the request
50			"""
51			include_fields = []
52			rif = request.get("include_fields", "")
53			if "include_fields" in request:
54			include_fields = [x.strip()
55			for x in rif.split(",")
56			if x.strip()]
57			if "include_fields[]" in request:
58			include_fields = request['include_fields[]']
59			return include_fields
60
61
62			def load_brain_metadata(proxy, include_fields):
63			"""Load values from the catalog metadata into a list of dictionaries
64			"""
65			ret = {}
66			for index in proxy.indexes():
67			if index not in proxy:
68			continue
69			if include_fields and index not in include_fields:
70			continue
71			val = getattr(proxy, index)
72			if val != Missing.Value:
73			try:
74			json.dumps(val)
75			except:
76			continue
77			ret[index] = val
78			return ret
79
80
81			def load_field_values(instance, include_fields):
82			"""Load values from an AT object schema fields into a list of dictionaries
83			"""
84			ret = {}
85			schema = instance.Schema()
86			val = None
87			for field in schema.fields():
88			fieldname = field.getName()
89			if include_fields and fieldname not in include_fields:
90			continue
91			try:
92			val = field.get(instance)
93			except AttributeError:
94			# If this error is raised, make a look to the add-on content
95			# expressions used to obtain their data.
96			print "AttributeError:", sys.exc_info()[1]
97			print "Unreachable object. Maybe the object comes from an Add-on"
98			print traceback.format_exc()
99
100			if val:
101			field_type = field.type
102			# If it a proxy field, we should know to the type of the proxied
103			# field
104			if field_type == 'proxy':
105			actual_field = field.get_proxy(instance)
106			field_type = actual_field.type
107			if field_type == "blob" or field_type == 'file':
108			continue
109			# I put the UID of all references here in *_uid.
110			if field_type == 'reference':
111			if type(val) in (list, tuple):
112			ret[fieldname + "_uid"] = [v.UID() for v in val]
113			val = [to_utf8(v.Title()) for v in val]
114			else:
115			ret[fieldname + "_uid"] = val.UID()
116			val = to_utf8(val.Title())
117			elif field_type == 'boolean':
118			val = True if val else False
119			elif field_type == 'text':
120			val = to_utf8(val)
121
122			try:
123			json.dumps(val)
124			except:
125			val = str(val)
126			ret[fieldname] = val
127			return ret
128
129
130			def get_include_methods(request):
131			"""Retrieve include_methods values from the request
132			"""
133			include_methods = request.get("include_methods[]")
134			if not include_methods:
135			include_methods = request.get("include_methods", [])
136
137			if isinstance(include_methods, six.string_types):
138			include_methods = include_methods.split(",")
139			include_methods = map(lambda me: me.strip(), include_methods)
140
141			return filter(None, include_methods)
142
143
144			def load_method_values(instance, include_methods):
145			ret = {}
146			for method in include_methods:
147			if hasattr(instance, method):
148			val = getattr(instance, method)()
149			ret[method] = val
150			return ret
151
152
153			def resolve_request_lookup(context, request, fieldname):
154			if fieldname not in request:
155			return []
156			brains = []
157			at = getToolByName(context, TOOL_NAME, None)
158			entries = request[fieldname] if type(request[fieldname]) in (list, tuple) \
159			else [request[fieldname], ]
160			for entry in entries:
161			contentFilter = {}
162			for value in entry.split("\|"):
163			if ":" in value:
164			index, value = value.split(":", 1)
165			else:
166			index, value = 'id', value
167			if index in contentFilter:
168			v = contentFilter[index]
169			v = v if type(v) in (list, tuple) else [v, ]
170			v.append(value)
171			contentFilter[index] = v
172			else:
173			contentFilter[index] = value
174			# search all possible catalogs
175			if 'portal_type' in contentFilter:
176			catalogs = at.getCatalogsByType(contentFilter['portal_type'])
177			elif 'uid' in contentFilter:
178			# If a uid is found, the object could be searched for its own uid
179			uc = getToolByName(context, 'uid_catalog')
180			return uc(UID=contentFilter['uid'])
181
182			else:
183			catalogs = [getToolByName(context, 'portal_catalog'), ]
184			for catalog in catalogs:
185			_brains = catalog(contentFilter)
186			if _brains:
187			brains.extend(_brains)
188			break
189			return brains
190
191
192			def set_fields_from_request(obj, request):
193			"""Search request for keys that match field names in obj,
194			and call field mutator with request value.
195
196			The list of fields for which schema mutators were found
197			is returned.
198			"""
199			schema = obj.Schema()
200			# fields contains all schema-valid field values from the request.
201			fields = {}
202			for fieldname, value in request.items():
203			if fieldname not in schema:
204			continue
205			field = schema[fieldname]
206			widget = field.getWidgetName()
207			if widget in ["ReferenceWidget"]:
208			brains = []
209			if value:
210			brains = resolve_request_lookup(obj, request, fieldname)
211			if not brains:
212			logger.warning(
213			"JSONAPI: Can't resolve reference: {} {}"
214			.format(fieldname, value))
215			return []
216			if schema[fieldname].multiValued:
217			value = [b.UID for b in brains] if brains else []
218			else:
219			value = brains[0].UID if brains else None
220			fields[fieldname] = value
221			# Write fields.
222			for fieldname, value in fields.items():
223			field = schema[fieldname]
224			fieldtype = field.getType()
225			if fieldtype == 'Products.Archetypes.Field.BooleanField':
226			if value.lower() in ('0', 'false', 'no') or not value:
227			value = False
228			else:
229			value = True
230			elif fieldtype in ['Products.ATExtensions.field.records.RecordsField',
231			'Products.ATExtensions.field.records.RecordField']:
232			try:
233			value = eval(value)
234			except:
235			logger.warning(
236			"JSONAPI: " + fieldname + ": Invalid "
237			"JSON/Python variable")
238			return []
239			mutator = field.getMutator(obj)
240			if mutator:
241			mutator(value)
242			else:
243			field.set(obj, value)
244			obj.reindexObject()
245			return fields.keys()
246

senaite / senaite.core

Push — master ( 2280ab...e4f397 )

bika.lims.jsonapi.set_fields_from_request() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like