How to use the jsonpatch.JsonPatchConflict function in jsonpatch

To help you get started, we’ve selected a few jsonpatch examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github EUDAT-B2SHARE / b2share / tests / b2share_unit_tests / schemas / test_schemas_api.py View on Github external
with app.app_context():
        created_community = Community.create_community(**community_metadata)
        block_schema = BlockSchema.create_block_schema(created_community.id, 'abc')
        block_schema_id = block_schema.id
        db.session.commit()

        retrieved = BlockSchema.get_block_schema(block_schema_id)
        retrieved.patch([{'op': 'replace', 'path': '/name', 'value': 'patched'}])
        db.session.commit()

    with app.app_context():
        patched = BlockSchema.get_block_schema(block_schema_id)
        assert block_schema_id == patched.id
        assert getattr(patched, 'name') == 'patched'

        with pytest.raises(JsonPatchConflict):
            patched.patch([{'op': 'replace', 'path': '/non_exist_name', 'value': None}])
        assert getattr(patched, 'name') == 'patched'
github EUDAT-B2SHARE / b2share / tests / b2share_unit_tests / communities / test_communities_api.py View on Github external
db.session.commit()

    with app.app_context():
        retrieved = Community.get(id=community_id)
        retrieved.patch(community_patch)
        db.session.commit()

    with app.app_context():
        patched = Community.get(id=community_id)
        assert community_id == patched.id
        assert patched.updated
        for field, value in patched_community_metadata.items():
            assert getattr(patched, field) == value

        # test invalid or conflicting patchs
        with pytest.raises(JsonPatchConflict):
            patched.patch([{
                'op': 'replace',
                'path': '/name',
                'value': 'this should not be applied'
            }, {
                'op': 'replace',
                'path': '/non-existing-field',
                'value': 'random value'
            }])
        with pytest.raises(InvalidJsonPatch):
            patched.patch({'whatever': 'key'})
        with pytest.raises(InvalidCommunityError):
            patched.patch([{
                'op': 'replace',
                'path': '/name',
                'value': None
github stefankoegl / python-json-patch / tests.py View on Github external
def test_replace_missing(self):
        src = {"foo": 1}
        patch_obj = [ { "op": "replace", "path": "/bar", "value": 10} ]
        self.assertRaises(jsonpatch.JsonPatchConflict, jsonpatch.apply_patch, src, patch_obj)
github CERNDocumentServer / cds-videos / cds / modules / webhooks / tasks.py View on Github external
def clean(self, deposit_id, version_id, *args, **kwargs):
        """Undo metadata extraction."""
        # 1. Revert patch on record
        recid = str(PersistentIdentifier.get(
            'depid', deposit_id).object_uuid)
        patch = [{
            'op': 'remove',
            'path': '/_cds/extracted_metadata',
        }]
        validator = 'cds.modules.records.validators.PartialDraft4Validator'
        try:
            patch_record(recid=recid, patch=patch, validator=validator)
        except jsonpatch.JsonPatchConflict as c:
            logger.warning(
                'Failed to apply JSON Patch to deposit {0}: {1}'.format(
                    recid, c))

        # Delete tmp file if any
        obj = as_object_version(version_id)
        temp_location = obj.get_tags().get('temp_location', None)
        if temp_location:
            shutil.rmtree(temp_location)
            ObjectVersionTag.delete(obj, 'temp_location')
            db.session.commit()
github stefankoegl / python-json-patch / jsonpatch.py View on Github external
raise InvalidJsonPatch(
                "The operation does not contain a 'value' member")

        subobj, part = self.pointer.to_last(obj)

        if part is None:
            return value

        if isinstance(subobj, MutableSequence):
            if part > len(subobj) or part < 0:
                raise JsonPatchConflict("can't replace outside of list")

        elif isinstance(subobj, MutableMapping):
            if part not in subobj:
                msg = "can't replace non-existent object '{0}'".format(part)
                raise JsonPatchConflict(msg)
        else:
            raise TypeError("invalid document type {0}".format(type(subobj)))

        subobj[part] = value
        return obj
github stefankoegl / python-json-patch / jsonpatch.py View on Github external
def apply(self, obj):
        try:
            value = self.operation["value"]
        except KeyError as ex:
            raise InvalidJsonPatch(
                "The operation does not contain a 'value' member")

        subobj, part = self.pointer.to_last(obj)

        if part is None:
            return value

        if isinstance(subobj, MutableSequence):
            if part > len(subobj) or part < 0:
                raise JsonPatchConflict("can't replace outside of list")

        elif isinstance(subobj, MutableMapping):
            if part not in subobj:
                msg = "can't replace non-existent object '{0}'".format(part)
                raise JsonPatchConflict(msg)
        else:
            raise TypeError("invalid document type {0}".format(type(subobj)))

        subobj[part] = value
        return obj
github stefankoegl / python-json-patch / jsonpatch.py View on Github external
def apply(self, obj):
        try:
            from_ptr = JsonPointer(self.operation['from'])
        except KeyError as ex:
            raise InvalidJsonPatch(
                "The operation does not contain a 'from' member")

        subobj, part = from_ptr.to_last(obj)
        try:
            value = copy.deepcopy(subobj[part])
        except (KeyError, IndexError) as ex:
            raise JsonPatchConflict(str(ex))

        obj = AddOperation({
            'op': 'add',
            'path': self.location,
            'value': value
        }).apply(obj)

        return obj
github openstack / solum / solum / api / controllers / camp / v1_1 / plans.py View on Github external
# check to make sure the request has the right Content-Type
        if (pecan.request.content_type is None or
                pecan.request.content_type != 'application/json-patch+json'):
            raise exception.UnsupportedMediaType(
                name=pecan.request.content_type,
                method='PATCH')

        try:
            patch = jsonpatch.JsonPatch.from_string(pecan.request.body)
            patched_obj = patch.apply(plan_obj.refined_content())
            db_obj = handler.update(uuid, patched_obj)
        except KeyError:
            # a key error indicates one of the patch operations is missing a
            # component
            raise exception.BadRequest(reason=MAL_PATCH_ERR)
        except jsonpatch.JsonPatchConflict:
            raise exception.Unprocessable
        except jsonpatch.JsonPatchException as jpe:
            raise JsonPatchProcessingException(reason=str(jpe))

        return fluff_plan(db_obj.refined_content(), db_obj.uuid)
github stefankoegl / python-json-patch / jsonpatch.py View on Github external
def apply(self, obj):
        try:
            value = self.operation["value"]
        except KeyError as ex:
            raise InvalidJsonPatch(
                "The operation does not contain a 'value' member")

        subobj, part = self.pointer.to_last(obj)

        if isinstance(subobj, MutableSequence):
            if part == '-':
                subobj.append(value)  # pylint: disable=E1103

            elif part > len(subobj) or part < 0:
                raise JsonPatchConflict("can't insert outside of list")

            else:
                subobj.insert(part, value)  # pylint: disable=E1103

        elif isinstance(subobj, MutableMapping):
            if part is None:
                obj = value  # we're replacing the root
            else:
                subobj[part] = value

        else:
            raise TypeError("invalid document type {0}".format(type(subobj)))

        return obj
github openstack / ironic / ironic / api / controllers / v1 / utils.py View on Github external
from ironic import api
from ironic.api.controllers.v1 import versions
from ironic.api import types as atypes
from ironic.common import exception
from ironic.common import faults
from ironic.common.i18n import _
from ironic.common import policy
from ironic.common import states
from ironic.common import utils
from ironic import objects


CONF = cfg.CONF


_JSONPATCH_EXCEPTIONS = (jsonpatch.JsonPatchConflict,
                         jsonpatch.JsonPatchException,
                         jsonpatch.JsonPointerException,
                         KeyError,
                         IndexError)


# Minimum API version to use for certain verbs
MIN_VERB_VERSIONS = {
    # v1.4 added the MANAGEABLE state and two verbs to move nodes into
    # and out of that state. Reject requests to do this in older versions
    states.VERBS['manage']: versions.MINOR_4_MANAGEABLE_STATE,
    states.VERBS['provide']: versions.MINOR_4_MANAGEABLE_STATE,

    states.VERBS['inspect']: versions.MINOR_6_INSPECT_STATE,
    states.VERBS['abort']: versions.MINOR_13_ABORT_VERB,
    states.VERBS['clean']: versions.MINOR_15_MANUAL_CLEAN,