"""Raised if patch could not be applied due to conflict situation such as:

"""Convert duplicate keys values to lists."""

# read all values into lists

mdict = collections.defaultdict(list)

for key, value in ordered_pairs:

mdict[key].append(value)

return dict(

# unpack lists that have only 1 item

(key, values[0] if len(values) == 1 else values)

for key, values in mdict.items()

"""Apply list of patches to specified json document.

if isinstance(patch, basestring):

patch = JsonPatch.from_string(patch, pointer_cls=pointer_cls)

else:

patch = JsonPatch(patch, pointer_cls=pointer_cls)

"""Generates patch by comparing two document objects. Actually is

"""A single operation inside a JSON Patch."""

def __init__(self, operation, pointer_cls=JsonPointer):

self.pointer_cls = pointer_cls

if not operation.__contains__('path'):

raise InvalidJsonPatch("Operation must have a 'path' member")

if isinstance(operation['path'], self.pointer_cls):

self.location = operation['path'].path

self.pointer = operation['path']

else:

self.location = operation['path']

try:

self.pointer = self.pointer_cls(self.location)

except TypeError as ex:

raise InvalidJsonPatch("Invalid 'path'")

self.operation = operation

def apply(self, obj):

"""Abstract method that applies a patch operation to the specified object."""

raise NotImplementedError('should implement the patch operation.')

def __hash__(self):

return hash(frozenset(self.operation.items()))

def __eq__(self, other):

if not isinstance(other, PatchOperation):

return False

return self.operation == other.operation

def __ne__(self, other):

return not(self == other)

@property

def path(self):

return '/'.join(self.pointer.parts[:-1])

@property

def key(self):

try:

return int(self.pointer.parts[-1])

except ValueError:

return self.pointer.parts[-1]

@key.setter

def key(self, value):

self.pointer.parts[-1] = str(value)

self.location = self.pointer.path

"""Removes an object property or an array element."""

def apply(self, obj):

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

try:

del subobj[part]

except (KeyError, IndexError) as ex:

msg = "can't remove a non-existent object '{0}'".format(part)

raise JsonPatchConflict(msg)

return obj

def _on_undo_remove(self, path, key):

if self.path == path:

if self.key >= key:

self.key += 1

else:

key -= 1

return key

def _on_undo_add(self, path, key):

if self.path == path:

if self.key > key:

self.key -= 1

else:

key -= 1

"""Adds an object property or an array element."""

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:

if part is None:

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

else:

raise JsonPatchConflict("unable to fully resolve json pointer {0}, part {1}".format(self.location, part))

return obj

def _on_undo_remove(self, path, key):

if self.path == path:

if self.key > key:

self.key += 1

else:

key += 1

return key

def _on_undo_add(self, path, key):

if self.path == path:

if self.key > key:

self.key -= 1

else:

key += 1

"""Replaces an object property or an array element by a new value."""

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 part == "-":

raise InvalidJsonPatch("'path' with '-' can't be applied to 'replace' operation")

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 a non-existent object '{0}'".format(part)

raise JsonPatchConflict(msg)

else:

if part is None:

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

else:

raise JsonPatchConflict("unable to fully resolve json pointer {0}, part {1}".format(self.location, part))

subobj[part] = value

return obj

def _on_undo_remove(self, path, key):

return key

def _on_undo_add(self, path, key):

"""Moves an object property or an array element to a new location."""

def apply(self, obj):

try:

if isinstance(self.operation['from'], self.pointer_cls):

from_ptr = self.operation['from']

else:

from_ptr = self.pointer_cls(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 = subobj[part]

except (KeyError, IndexError) as ex:

raise JsonPatchConflict(str(ex))

# If source and target are equal, this is a no-op

if self.pointer == from_ptr:

return obj

if isinstance(subobj, MutableMapping) and \

self.pointer.contains(from_ptr):

raise JsonPatchConflict('Cannot move values into their own children')

obj = RemoveOperation({

'op': 'remove',

'path': self.operation['from']

}, pointer_cls=self.pointer_cls).apply(obj)

obj = AddOperation({

'op': 'add',

'path': self.location,

'value': value

}, pointer_cls=self.pointer_cls).apply(obj)

return obj

@property

def from_path(self):

from_ptr = self.pointer_cls(self.operation['from'])

return '/'.join(from_ptr.parts[:-1])

@property

def from_key(self):

from_ptr = self.pointer_cls(self.operation['from'])

try:

return int(from_ptr.parts[-1])

except TypeError:

return from_ptr.parts[-1]

@from_key.setter

def from_key(self, value):

from_ptr = self.pointer_cls(self.operation['from'])

from_ptr.parts[-1] = str(value)

self.operation['from'] = from_ptr.path

def _on_undo_remove(self, path, key):

if self.from_path == path:

if self.from_key >= key:

self.from_key += 1

else:

key -= 1

if self.path == path:

if self.key > key:

self.key += 1

else:

key += 1

return key

def _on_undo_add(self, path, key):

if self.from_path == path:

if self.from_key > key:

self.from_key -= 1

else:

key -= 1

if self.path == path:

if self.key > key:

self.key -= 1

else:

key += 1

"""Test value by specified location."""

def apply(self, obj):

try:

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

if part is None:

val = subobj

else:

val = self.pointer.walk(subobj, part)

except JsonPointerException as ex:

raise JsonPatchTestFailed(str(ex))

try:

value = self.operation['value']

except KeyError as ex:

raise InvalidJsonPatch(

"The operation does not contain a 'value' member")

if val != value:

msg = '{0} ({1}) is not equal to tested value {2} ({3})'

raise JsonPatchTestFailed(msg.format(val, type(val),

value, type(value)))

""" Copies an object property or an array element to a new location """

def apply(self, obj):

try:

from_ptr = self.pointer_cls(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

}, pointer_cls=self.pointer_cls).apply(obj)

json_dumper = staticmethod(json.dumps)

json_loader = staticmethod(_jsonloads)

operations = MappingProxyType({

'remove': RemoveOperation,

'add': AddOperation,

'replace': ReplaceOperation,

'move': MoveOperation,

'test': TestOperation,

'copy': CopyOperation,

})

"""A JSON Patch is a list of Patch Operations.

def __init__(self, patch, pointer_cls=JsonPointer):

self.patch = patch

self.pointer_cls = pointer_cls

# Verify that the structure of the patch document

# is correct by retrieving each patch element.

# Much of the validation is done in the initializer

# though some is delayed until the patch is applied.

for op in self.patch:

self._get_operation(op)

def __str__(self):

"""str(self) -> self.to_string()"""

return self.to_string()

def __bool__(self):

return bool(self.patch)

__nonzero__ = __bool__

def __iter__(self):

return iter(self.patch)

def __hash__(self):

return hash(tuple(self._ops))

def __eq__(self, other):

if not isinstance(other, JsonPatch):

return False

return self._ops == other._ops

def __ne__(self, other):

return not(self == other)

@classmethod

def from_string(cls, patch_str, loads=None, pointer_cls=JsonPointer):

"""Creates JsonPatch instance from string source.

json_loader = loads or cls.json_loader

patch = json_loader(patch_str)

return cls(patch, pointer_cls=pointer_cls)

@classmethod

def from_diff(

cls, src, dst, optimization=True, dumps=None,

pointer_cls=JsonPointer,

):

"""Creates JsonPatch instance based on comparison of two document

json_dumper = dumps or cls.json_dumper

builder = DiffBuilder(src, dst, json_dumper, pointer_cls=pointer_cls)

builder._compare_values('', None, src, dst)

ops = list(builder.execute())

return cls(ops, pointer_cls=pointer_cls)

def to_string(self, dumps=None):

"""Returns patch set as JSON string."""

json_dumper = dumps or self.json_dumper

return json_dumper(self.patch)

@property

def _ops(self):

return tuple(map(self._get_operation, self.patch))

def apply(self, obj, in_place=False):

"""Applies the patch to a given object.

if not in_place:

obj = copy.deepcopy(obj)

for operation in self._ops:

obj = operation.apply(obj)

return obj

def _get_operation(self, operation):

if 'op' not in operation:

raise InvalidJsonPatch("Operation does not contain 'op' member")

op = operation['op']

if not isinstance(op, basestring):

raise InvalidJsonPatch("Operation must be a string")

if op not in self.operations:

raise InvalidJsonPatch("Unknown operation {0!r}".format(op))

cls = self.operations[op]

def __init__(self, src_doc, dst_doc, dumps=json.dumps, pointer_cls=JsonPointer):

self.dumps = dumps

self.pointer_cls = pointer_cls

self.index_storage = [{}, {}]

self.index_storage2 = [[], []]

self.__root = root = []

self.src_doc = src_doc

self.dst_doc = dst_doc

root[:] = [root, root, None]

def store_index(self, value, index, st):

typed_key = (value, type(value))

try:

storage = self.index_storage[st]

stored = storage.get(typed_key)

if stored is None:

storage[typed_key] = [index]

else:

storage[typed_key].append(index)

except TypeError:

self.index_storage2[st].append((typed_key, index))

def take_index(self, value, st):

typed_key = (value, type(value))

try:

stored = self.index_storage[st].get(typed_key)

if stored:

return stored.pop()

except TypeError:

storage = self.index_storage2[st]

for i in range(len(storage)-1, -1, -1):

if storage[i][0] == typed_key:

return storage.pop(i)[1]

def insert(self, op):

root = self.__root

last = root[0]

last[1] = root[0] = [last, root, op]

return root[0]

def remove(self, index):

link_prev, link_next, _ = index

link_prev[1] = link_next

link_next[0] = link_prev

index[:] = []

def iter_from(self, start):

root = self.__root

curr = start[1]

while curr is not root:

yield curr[2]

curr = curr[1]

def __iter__(self):

root = self.__root

curr = root[1]

while curr is not root:

yield curr[2]

curr = curr[1]

def execute(self):

root = self.__root

curr = root[1]

while curr is not root:

if curr[1] is not root:

op_first, op_second = curr[2], curr[1][2]

if op_first.location == op_second.location and \

type(op_first) == RemoveOperation and \

type(op_second) == AddOperation:

yield ReplaceOperation({

'op': 'replace',

'path': op_second.location,

'value': op_second.operation['value'],

}, pointer_cls=self.pointer_cls).operation

curr = curr[1][1]

continue

yield curr[2].operation

curr = curr[1]

def _item_added(self, path, key, item):

index = self.take_index(item, _ST_REMOVE)

if index is not None:

op = index[2]

if type(op.key) == int and type(key) == int:

for v in self.iter_from(index):

op.key = v._on_undo_remove(op.path, op.key)

self.remove(index)

if op.location != _path_join(path, key):

new_op = MoveOperation({

'op': 'move',

'from': op.location,

'path': _path_join(path, key),

}, pointer_cls=self.pointer_cls)

self.insert(new_op)

else:

new_op = AddOperation({

'op': 'add',

'path': _path_join(path, key),

'value': item,

}, pointer_cls=self.pointer_cls)

new_index = self.insert(new_op)

self.store_index(item, new_index, _ST_ADD)

def _item_removed(self, path, key, item):

new_op = RemoveOperation({

'op': 'remove',

'path': _path_join(path, key),

}, pointer_cls=self.pointer_cls)

index = self.take_index(item, _ST_ADD)

new_index = self.insert(new_op)

if index is not None:

op = index[2]

# We can't rely on the op.key type since PatchOperation casts

# the .key property to int and this path wrongly ends up being taken

# for numeric string dict keys while the intention is to only handle lists.

# So we do an explicit check on the item affected by the op instead.

added_item = op.pointer.to_last(self.dst_doc)[0]

if type(added_item) == list:

for v in self.iter_from(index):

op.key = v._on_undo_add(op.path, op.key)

self.remove(index)

if new_op.location != op.location:

new_op = MoveOperation({

'op': 'move',

'from': new_op.location,

'path': op.location,

}, pointer_cls=self.pointer_cls)

new_index[2] = new_op

else:

self.remove(new_index)

else:

self.store_index(item, new_index, _ST_REMOVE)

def _item_replaced(self, path, key, item):

self.insert(ReplaceOperation({

'op': 'replace',

'path': _path_join(path, key),

'value': item,

}, pointer_cls=self.pointer_cls))

def _compare_dicts(self, path, src, dst):

src_keys = set(src.keys())

dst_keys = set(dst.keys())

added_keys = dst_keys - src_keys

removed_keys = src_keys - dst_keys

for key in removed_keys:

self._item_removed(path, str(key), src[key])

for key in added_keys:

self._item_added(path, str(key), dst[key])

for key in src_keys & dst_keys:

self._compare_values(path, key, src[key], dst[key])

def _compare_lists(self, path, src, dst):

len_src, len_dst = len(src), len(dst)

max_len = max(len_src, len_dst)

min_len = min(len_src, len_dst)

for key in range(max_len):

if key < min_len:

old, new = src[key], dst[key]

if old == new:

continue

elif isinstance(old, MutableMapping) and \

isinstance(new, MutableMapping):

self._compare_dicts(_path_join(path, key), old, new)

elif isinstance(old, MutableSequence) and \

isinstance(new, MutableSequence):

self._compare_lists(_path_join(path, key), old, new)

else:

self._item_removed(path, key, old)

self._item_added(path, key, new)

elif len_src > len_dst:

self._item_removed(path, len_dst, src[key])

else:

self._item_added(path, key, dst[key])

def _compare_values(self, path, key, src, dst):

if isinstance(src, MutableMapping) and \

isinstance(dst, MutableMapping):

self._compare_dicts(_path_join(path, key), src, dst)

elif isinstance(src, MutableSequence) and \

isinstance(dst, MutableSequence):

self._compare_lists(_path_join(path, key), src, dst)

# To ensure we catch changes to JSON, we can't rely on a simple

# src == dst, because it would not recognize the difference between

# 1 and True, among other things. Using json.dumps is the most

# fool-proof way to ensure we catch type changes that matter to JSON

# and ignore those that don't. The performance of this could be

# improved by doing more direct type checks, but we'd need to be

# careful to accept type changes that don't matter when JSONified.

elif self.dumps(src) == self.dumps(dst):

return

else:

if key is None:

return path