Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import ast
import jsonpatch
from oslo_utils import uuidutils
import pecan
import wsme
from magnum.common import exception
from magnum.common import utils
import magnum.conf
from magnum.i18n import _
from magnum import objects
CONF = magnum.conf.CONF
JSONPATCH_EXCEPTIONS = (jsonpatch.JsonPatchException,
jsonpatch.JsonPointerException,
KeyError)
DOCKER_MINIMUM_MEMORY = 4 * 1024 * 1024
def validate_limit(limit):
if limit is not None and limit <= 0:
raise wsme.exc.ClientSideError(_("Limit must be positive"))
if limit is not None:
return min(CONF.api.max_limit, limit)
else:
return CONF.api.max_limit
def partial_update(self, request, *args, **kwargs):
patch = JsonPatch(request.DATA)
obj = self.get_object()
serializer = self.get_serializer(instance=obj)
doc = serializer.data
try:
# `jsonpatch` does not force documents to be array of operations
# So we have to do it manually
if not isinstance(request.DATA, list):
raise JsonPatchException(
"The patch must be supplied as a list",
)
modified = patch.apply(doc)
# Set the modified data to the request data
# This will allow us to update the object using it
request._data = modified
return super(JsonPatchMixin, self).update(request, *args, **kwargs)
except JsonPatchException as ex:
message = force_text(ex)
# `jsonpatch` does not handle unicode transparently
# So we have to strip out the `u'` in Python 2
@app.errorhandler(JsonPatchException)
def handle_validation_error(err):
return InvalidOperationError(
errors=[GenericError('JSON-Patch error: {}'.format(err.args[0]))]
)
__author__ = 'Stefan Kögl '
__version__ = '1.21'
__website__ = 'https://github.com/stefankoegl/python-json-patch'
__license__ = 'Modified BSD License'
# pylint: disable=E0611,W0404
if sys.version_info >= (3, 0):
basestring = (bytes, str) # pylint: disable=C0103,W0622
class JsonPatchException(Exception):
"""Base Json Patch exception"""
class InvalidJsonPatch(JsonPatchException):
""" Raised if an invalid JSON Patch is created """
class JsonPatchConflict(JsonPatchException):
"""Raised if patch could not be applied due to conflict situation such as:
- attempt to add object key then it already exists;
- attempt to operate with nonexistence object key;
- attempt to insert value to array at position beyond of it size;
- etc.
"""
class JsonPatchTestFailed(JsonPatchException, AssertionError):
""" A Test operation failed """
# pylint: disable=E0611,W0404
if sys.version_info >= (3, 0):
basestring = (bytes, str) # pylint: disable=C0103,W0622
class JsonPatchException(Exception):
"""Base Json Patch exception"""
class InvalidJsonPatch(JsonPatchException):
""" Raised if an invalid JSON Patch is created """
class JsonPatchConflict(JsonPatchException):
"""Raised if patch could not be applied due to conflict situation such as:
- attempt to add object key then it already exists;
- attempt to operate with nonexistence object key;
- attempt to insert value to array at position beyond of it size;
- etc.
"""
class JsonPatchTestFailed(JsonPatchException, AssertionError):
""" A Test operation failed """
def multidict(ordered_pairs):
"""Convert duplicate keys values to lists."""
# read all values into lists
mdict = collections.defaultdict(list)
try:
rt = pecan.request.indexer.get_resource_type(self._name)
except indexer.NoSuchResourceType as e:
abort(404, six.text_type(e))
enforce("update resource type", rt)
# Ensure this is a valid jsonpatch dict
patch = deserialize_and_validate(
ResourceTypeJsonPatchSchema,
expected_content_types=["application/json-patch+json"])
# Add new attributes to the resource type
rt_json_current = rt.jsonify()
try:
rt_json_next = jsonpatch.apply_patch(rt_json_current, patch)
except jsonpatch.JsonPatchException as e:
abort(400, six.text_type(e))
del rt_json_next['state']
# Validate that the whole new resource_type is valid
schema = pecan.request.indexer.get_resource_type_schema()
try:
rt_json_next = voluptuous.Schema(schema.for_update, required=True)(
rt_json_next)
except voluptuous.Error as e:
abort(400, "Invalid input: %s" % e)
# Get only newly formatted and deleted attributes
add_attrs = {k: v for k, v in rt_json_next["attributes"].items()
if k not in rt_json_current["attributes"]}
del_attrs = [k for k in rt_json_current["attributes"]
if k not in rt_json_next["attributes"]]
# `jsonpatch` does not force documents to be array of operations
# So we have to do it manually
if not isinstance(request.DATA, list):
raise JsonPatchException(
"The patch must be supplied as a list",
)
modified = patch.apply(doc)
# Set the modified data to the request data
# This will allow us to update the object using it
request._data = modified
return super(JsonPatchMixin, self).update(request, *args, **kwargs)
except JsonPatchException as ex:
message = force_text(ex)
# `jsonpatch` does not handle unicode transparently
# So we have to strip out the `u'` in Python 2
if "Unknown operation u'" in message and sys.version_info < (3, 0):
message = message.replace("u'", "'")
data = {
"detail": message,
}
return response.Response(data, status=400)
def patch(self, request, *args, **kwargs):
patch = request.DATA
obj = self.get_object()
serializer = self.get_serializer(instance=obj,
context={'request': request,
'view': self})
try:
new_data = patch.apply(serializer.data)
except (jsonpatch.JsonPatchException,
jsonpointer.JsonPointerException) as e:
msg = 'Could not apply json-patch to object: {}'.format(e)
raise exceptions.BadUpdate(detail=msg)
ret_val = self.update_object(request, new_data)
return ret_val