bika.lims.jsonapi.create.Create.require() - 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

bika.lims.jsonapi.create.Create.require() B

↳ Parent: bika.lims.jsonapi.create

Complexity

Conditions

Size

Total Lines	6
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	5
dl	0
loc	6
rs	8.6666
c	0
b	0
f	0
cc	6
nop	3

# -*- 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 AccessControl import getSecurityManager
from AccessControl import Unauthorized
from bika.lims.idserver import renameAfterCreation
from bika.lims.jsonapi import set_fields_from_request
from bika.lims.permissions import AccessJSONAPI
from bika.lims.utils import tmpID
from plone.jsonapi.core import router
from plone.jsonapi.core.interfaces import IRouteProvider
from Products.Archetypes.event import ObjectInitializedEvent
from Products.CMFPlone.utils import _createObjectByType
from zExceptions import BadRequest
from zope import event
from zope import interface

import transaction

class Create(object):
    interface.implements(IRouteProvider)

    def initialize(self, context, request):
        self.context = context
        self.request = request

    @property
    def routes(self):
        return (
            ("/create", "create", self.create, dict(methods=['GET', 'POST'])),
        )


    def create(self, context, request):
        """/@@API/create: Create new object.

        Required parameters:

            - obj_type = portal_type of new object.
            - obj_path = path of new object, from plone site root. - Not required for
             obj_type=AnalysisRequest

        Optionally:

            - obj_id = ID of new object.

        All other parameters in the request are matched against the object's
        Schema.  If a matching field is found in the schema, then the value is
        taken from the request and sent to the field's mutator.

        Reference fields may have their target value(s) specified with a
        delimited string query syntax, containing the portal_catalog search:

            <FieldName>=index1:value1|index2:value2

        eg to set the Client of a batch:

            ...@@API/update?obj_path=<path>...
            ...&Client=title:<client_title>&...

        And, to set a multi-valued reference, these both work:

            ...@@API/update?obj_path=<path>...
            ...&InheritedObjects:list=title:AR1...
            ...&InheritedObjects:list=title:AR2...

            ...@@API/update?obj_path=<path>...
            ...&InheritedObjects[]=title:AR1...
            ...&InheritedObjects[]=title:AR2...

        The Analysis_Specification parameter is special, it mimics
        the format of the python dictionaries, and only service Keyword
        can be used to reference services.  Even if the keyword is not
        actively required, it must be supplied:

            <service_keyword>:min:max:error tolerance

        The function returns a dictionary as a json string:

        {
            runtime: Function running time.
            error: true or string(message) if error. false if no error.
            success: true or string(message) if success. false if no success.
        }

        >>> portal = layer['portal']
        >>> portal_url = portal.absolute_url()
        >>> from plone.app.testing import SITE_OWNER_NAME
        >>> from plone.app.testing import SITE_OWNER_PASSWORD

        Simple AR creation, no obj_path parameter is required:

        >>> browser = layer['getBrowser'](portal, loggedIn=True, username=SITE_OWNER_NAME, password=SITE_OWNER_PASSWORD)
        >>> browser.open(portal_url+"/@@API/create", "&".join([
        ... "obj_type=AnalysisRequest",
        ... "Client=portal_type:Client|id:client-1",
        ... "SampleType=portal_type:SampleType|title:Apple Pulp",
        ... "Contact=portal_type:Contact|getFullname:Rita Mohale",
        ... "Services:list=portal_type:AnalysisService|title:Calcium",
        ... "Services:list=portal_type:AnalysisService|title:Copper",
        ... "Services:list=portal_type:AnalysisService|title:Magnesium",
        ... "SamplingDate=2013-09-29",
        ... "Specification=portal_type:AnalysisSpec|title:Apple Pulp",
        ... ]))
        >>> browser.contents
        '{..."success": true...}'

        If some parameters are specified and are not located as existing fields or properties
        of the created instance, the create should fail:

        >>> browser = layer['getBrowser'](portal, loggedIn=True, username=SITE_OWNER_NAME, password=SITE_OWNER_PASSWORD)
        >>> browser.open(portal_url+"/@@API/create?", "&".join([
        ... "obj_type=Batch",
        ... "obj_path=/batches",
        ... "title=Test",
        ... "Thing=Fish"
        ... ]))
        >>> browser.contents
        '{...The following request fields were not used: ...Thing...}'

        Now we test that the AR create also fails if some fields are spelled wrong

        >>> browser = layer['getBrowser'](portal, loggedIn=True, username=SITE_OWNER_NAME, password=SITE_OWNER_PASSWORD)
        >>> browser.open(portal_url+"/@@API/create", "&".join([
        ... "obj_type=AnalysisRequest",
        ... "thing=Fish",
        ... "Client=portal_type:Client|id:client-1",
        ... "SampleType=portal_type:SampleType|title:Apple Pulp",
        ... "Contact=portal_type:Contact|getFullname:Rita Mohale",
        ... "Services:list=portal_type:AnalysisService|title:Calcium",
        ... "Services:list=portal_type:AnalysisService|title:Copper",
        ... "Services:list=portal_type:AnalysisService|title:Magnesium",
        ... "SamplingDate=2013-09-29"
        ... ]))
        >>> browser.contents
        '{...The following request fields were not used: ...thing...}'

        """
        savepoint = transaction.savepoint()
        self.context = context
        self.request = request
        self.unused = [x for x in self.request.form.keys()]
        self.used("form.submitted")
        self.used("__ac_name")
        self.used("__ac_password")
        # always require obj_type
        self.require("obj_type")
        obj_type = self.request['obj_type']
        self.used("obj_type")
        # AnalysisRequest shortcut: creates Sample, Partition, AR, Analyses.
        if obj_type == "AnalysisRequest":
            raise BadRequest("Creation of Analysis Request through JSON API is "
                             "not supported. Request aborted.")
        # Other object types require explicit path as their parent
        self.require("obj_path")
        obj_path = self.request['obj_path']
        if not obj_path.startswith("/"):
            obj_path = "/" + obj_path
        self.used("obj_path")
        site_path = request['PATH_INFO'].replace("/@@API/create", "")
        parent = context.restrictedTraverse(str(site_path + obj_path))
        # normal permissions still apply for this user
        if not getSecurityManager().checkPermission(AccessJSONAPI, parent):
            msg = "You don't have the '{0}' permission on {1}".format(
                AccessJSONAPI, parent.absolute_url())
            raise Unauthorized(msg)

        obj_id = request.get("obj_id", "")
        _renameAfterCreation = False
        if not obj_id:
            _renameAfterCreation = True
            obj_id = tmpID()
        self.used(obj_id)

        ret = {
            "url": router.url_for("create", force_external=True),
            "success": True,
            "error": False,
        }

        try:
            obj = _createObjectByType(obj_type, parent, obj_id)
            obj.unmarkCreationFlag()
            if _renameAfterCreation:
                renameAfterCreation(obj)
            ret['obj_id'] = obj.getId()
            ret['obj_uid'] = obj.UID()
            used_fields = set_fields_from_request(obj, request)
            for field in used_fields:
                self.used(field)
            obj.reindexObject()
            obj.aq_parent.reindexObject()
            event.notify(ObjectInitializedEvent(obj))
            obj.at_post_create_script()
        except:
            savepoint.rollback()
            raise

        if self.unused:
            raise BadRequest("The following request fields were not used: %s.  Request aborted." % self.unused)

        return ret

    def require(self, fieldname, allow_blank=False):
        """fieldname is required"""
        if self.request.form and fieldname not in self.request.form.keys():
            raise Exception("Required field not found in request: %s" % fieldname)
        if self.request.form and (not self.request.form[fieldname] or allow_blank):
            raise Exception("Required field %s may not have blank value")

    def used(self, fieldname):
        """fieldname is used, remove from list of unused fields"""
        if fieldname in self.unused:
            self.unused.remove(fieldname)


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 AccessControl import getSecurityManager
9			from AccessControl import Unauthorized
10			from bika.lims.idserver import renameAfterCreation
11			from bika.lims.jsonapi import set_fields_from_request
12			from bika.lims.permissions import AccessJSONAPI
13			from bika.lims.utils import tmpID
14			from plone.jsonapi.core import router
15			from plone.jsonapi.core.interfaces import IRouteProvider
16			from Products.Archetypes.event import ObjectInitializedEvent
17			from Products.CMFPlone.utils import _createObjectByType
18			from zExceptions import BadRequest
19			from zope import event
20			from zope import interface
21
22			import transaction
23
24			class Create(object):
25			interface.implements(IRouteProvider)
26
27			def initialize(self, context, request):
28			self.context = context
29			self.request = request
30
31			@property
32			def routes(self):
33			return (
34			("/create", "create", self.create, dict(methods=['GET', 'POST'])),
35			)
36
37
38			def create(self, context, request):
39			"""/@@API/create: Create new object.
40
41			Required parameters:
42
43			- obj_type = portal_type of new object.
44			- obj_path = path of new object, from plone site root. - Not required for
45			obj_type=AnalysisRequest
46
47			Optionally:
48
49			- obj_id = ID of new object.
50
51			All other parameters in the request are matched against the object's
52			Schema. If a matching field is found in the schema, then the value is
53			taken from the request and sent to the field's mutator.
54
55			Reference fields may have their target value(s) specified with a
56			delimited string query syntax, containing the portal_catalog search:
57
58			<FieldName>=index1:value1\|index2:value2
59
60			eg to set the Client of a batch:
61
62			...@@API/update?obj_path=<path>...
63			...&Client=title:<client_title>&...
64
65			And, to set a multi-valued reference, these both work:
66
67			...@@API/update?obj_path=<path>...
68			...&InheritedObjects:list=title:AR1...
69			...&InheritedObjects:list=title:AR2...
70
71			...@@API/update?obj_path=<path>...
72			...&InheritedObjects[]=title:AR1...
73			...&InheritedObjects[]=title:AR2...
74
75			The Analysis_Specification parameter is special, it mimics
76			the format of the python dictionaries, and only service Keyword
77			can be used to reference services. Even if the keyword is not
78			actively required, it must be supplied:
79
80			<service_keyword>:min:max:error tolerance
81
82			The function returns a dictionary as a json string:
83
84			{
85			runtime: Function running time.
86			error: true or string(message) if error. false if no error.
87			success: true or string(message) if success. false if no success.
88			}
89
90			>>> portal = layer['portal']
91			>>> portal_url = portal.absolute_url()
92			>>> from plone.app.testing import SITE_OWNER_NAME
93			>>> from plone.app.testing import SITE_OWNER_PASSWORD
94
95			Simple AR creation, no obj_path parameter is required:
96
97			>>> browser = layer['getBrowser'](portal, loggedIn=True, username=SITE_OWNER_NAME, password=SITE_OWNER_PASSWORD)
98			>>> browser.open(portal_url+"/@@API/create", "&".join([
99			... "obj_type=AnalysisRequest",
100			... "Client=portal_type:Client\|id:client-1",
101			... "SampleType=portal_type:SampleType\|title:Apple Pulp",
102			... "Contact=portal_type:Contact\|getFullname:Rita Mohale",
103			... "Services:list=portal_type:AnalysisService\|title:Calcium",
104			... "Services:list=portal_type:AnalysisService\|title:Copper",
105			... "Services:list=portal_type:AnalysisService\|title:Magnesium",
106			... "SamplingDate=2013-09-29",
107			... "Specification=portal_type:AnalysisSpec\|title:Apple Pulp",
108			... ]))
109			>>> browser.contents
110			'{..."success": true...}'
111
112			If some parameters are specified and are not located as existing fields or properties
113			of the created instance, the create should fail:
114
115			>>> browser = layer['getBrowser'](portal, loggedIn=True, username=SITE_OWNER_NAME, password=SITE_OWNER_PASSWORD)
116			>>> browser.open(portal_url+"/@@API/create?", "&".join([
117			... "obj_type=Batch",
118			... "obj_path=/batches",
119			... "title=Test",
120			... "Thing=Fish"
121			... ]))
122			>>> browser.contents
123			'{...The following request fields were not used: ...Thing...}'
124
125			Now we test that the AR create also fails if some fields are spelled wrong
126
127			>>> browser = layer['getBrowser'](portal, loggedIn=True, username=SITE_OWNER_NAME, password=SITE_OWNER_PASSWORD)
128			>>> browser.open(portal_url+"/@@API/create", "&".join([
129			... "obj_type=AnalysisRequest",
130			... "thing=Fish",
131			... "Client=portal_type:Client\|id:client-1",
132			... "SampleType=portal_type:SampleType\|title:Apple Pulp",
133			... "Contact=portal_type:Contact\|getFullname:Rita Mohale",
134			... "Services:list=portal_type:AnalysisService\|title:Calcium",
135			... "Services:list=portal_type:AnalysisService\|title:Copper",
136			... "Services:list=portal_type:AnalysisService\|title:Magnesium",
137			... "SamplingDate=2013-09-29"
138			... ]))
139			>>> browser.contents
140			'{...The following request fields were not used: ...thing...}'
141
142			"""
143			savepoint = transaction.savepoint()
144			self.context = context
145			self.request = request
146			self.unused = [x for x in self.request.form.keys()]
147			self.used("form.submitted")
148			self.used("__ac_name")
149			self.used("__ac_password")
150			# always require obj_type
151			self.require("obj_type")
152			obj_type = self.request['obj_type']
153			self.used("obj_type")
154			# AnalysisRequest shortcut: creates Sample, Partition, AR, Analyses.
155			if obj_type == "AnalysisRequest":
156			raise BadRequest("Creation of Analysis Request through JSON API is "
157			"not supported. Request aborted.")
158			# Other object types require explicit path as their parent
159			self.require("obj_path")
160			obj_path = self.request['obj_path']
161			if not obj_path.startswith("/"):
162			obj_path = "/" + obj_path
163			self.used("obj_path")
164			site_path = request['PATH_INFO'].replace("/@@API/create", "")
165			parent = context.restrictedTraverse(str(site_path + obj_path))
166			# normal permissions still apply for this user
167			if not getSecurityManager().checkPermission(AccessJSONAPI, parent):
168			msg = "You don't have the '{0}' permission on {1}".format(
169			AccessJSONAPI, parent.absolute_url())
170			raise Unauthorized(msg)
171
172			obj_id = request.get("obj_id", "")
173			_renameAfterCreation = False
174			if not obj_id:
175			_renameAfterCreation = True
176			obj_id = tmpID()
177			self.used(obj_id)
178
179			ret = {
180			"url": router.url_for("create", force_external=True),
181			"success": True,
182			"error": False,
183			}
184
185			try:
186			obj = _createObjectByType(obj_type, parent, obj_id)
187			obj.unmarkCreationFlag()
188			if _renameAfterCreation:
189			renameAfterCreation(obj)
190			ret['obj_id'] = obj.getId()
191			ret['obj_uid'] = obj.UID()
192			used_fields = set_fields_from_request(obj, request)
193			for field in used_fields:
194			self.used(field)
195			obj.reindexObject()
196			obj.aq_parent.reindexObject()
197			event.notify(ObjectInitializedEvent(obj))
198			obj.at_post_create_script()
199			except:
200			savepoint.rollback()
201			raise
202
203			if self.unused:
204			raise BadRequest("The following request fields were not used: %s. Request aborted." % self.unused)
205
206			return ret
207
208			def require(self, fieldname, allow_blank=False):
209			"""fieldname is required"""
210			if self.request.form and fieldname not in self.request.form.keys():
211			raise Exception("Required field not found in request: %s" % fieldname)
212			if self.request.form and (not self.request.form[fieldname] or allow_blank):
213			raise Exception("Required field %s may not have blank value")
214
215			def used(self, fieldname):
216			"""fieldname is used, remove from list of unused fields"""
217			if fieldname in self.unused:
218			self.unused.remove(fieldname)
219

senaite / senaite.core

Branch — master (9edffc)

bika.lims.jsonapi.create.Create.require() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like