From b8083d703c3aacf52429a06dc5b482a1f9acf54f Mon Sep 17 00:00:00 2001 From: Artyom Nikitin Date: Mon, 23 Nov 2020 23:52:42 +0300 Subject: feat: make operations class-based --- jsonpatch.py | 395 ++++++++++++++++++++++++++++++----------------------------- 1 file changed, 200 insertions(+), 195 deletions(-) diff --git a/jsonpatch.py b/jsonpatch.py index 5522d50..a01a177 100644 --- a/jsonpatch.py +++ b/jsonpatch.py @@ -39,6 +39,11 @@ import copy import functools import json import sys +try: + from types import MappingProxyType +except ImportError: + # Python < 3.3 + MappingProxyType = dict from jsonpointer import JsonPointer, JsonPointerException @@ -170,201 +175,6 @@ def make_patch(src, dst, pointer_cls=JsonPointer): return JsonPatch.from_diff(src, dst, pointer_cls=pointer_cls) -class JsonPatch(object): - json_dumper = staticmethod(json.dumps) - json_loader = staticmethod(_jsonloads) - - """A JSON Patch is a list of Patch Operations. - - >>> patch = JsonPatch([ - ... {'op': 'add', 'path': '/foo', 'value': 'bar'}, - ... {'op': 'add', 'path': '/baz', 'value': [1, 2, 3]}, - ... {'op': 'remove', 'path': '/baz/1'}, - ... {'op': 'test', 'path': '/baz', 'value': [1, 3]}, - ... {'op': 'replace', 'path': '/baz/0', 'value': 42}, - ... {'op': 'remove', 'path': '/baz/1'}, - ... ]) - >>> doc = {} - >>> result = patch.apply(doc) - >>> expected = {'foo': 'bar', 'baz': [42]} - >>> result == expected - True - - JsonPatch object is iterable, so you can easily access each patch - statement in a loop: - - >>> lpatch = list(patch) - >>> expected = {'op': 'add', 'path': '/foo', 'value': 'bar'} - >>> lpatch[0] == expected - True - >>> lpatch == patch.patch - True - - Also JsonPatch could be converted directly to :class:`bool` if it contains - any operation statements: - - >>> bool(patch) - True - >>> bool(JsonPatch([])) - False - - This behavior is very handy with :func:`make_patch` to write more readable - code: - - >>> old = {'foo': 'bar', 'numbers': [1, 3, 4, 8]} - >>> new = {'baz': 'qux', 'numbers': [1, 4, 7]} - >>> patch = make_patch(old, new) - >>> if patch: - ... # document have changed, do something useful - ... patch.apply(old) #doctest: +ELLIPSIS - {...} - """ - def __init__(self, patch, pointer_cls=JsonPointer): - self.patch = patch - self.pointer_cls = pointer_cls - - self.operations = { - 'remove': RemoveOperation, - 'add': AddOperation, - 'replace': ReplaceOperation, - 'move': MoveOperation, - 'test': TestOperation, - 'copy': CopyOperation, - } - - # 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. - - :param patch_str: JSON patch as raw string. - :type patch_str: str - - :param loads: A function of one argument that loads a serialized - JSON string. - :type loads: function - - :param pointer_cls: JSON pointer class to use. - :type pointer_cls: Type[JsonPointer] - - :return: :class:`JsonPatch` instance. - """ - 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 - objects. Json patch would be created for `src` argument against `dst` - one. - - :param src: Data source document object. - :type src: dict - - :param dst: Data source document object. - :type dst: dict - - :param dumps: A function of one argument that produces a serialized - JSON string. - :type dumps: function - - :param pointer_cls: JSON pointer class to use. - :type pointer_cls: Type[JsonPointer] - - :return: :class:`JsonPatch` instance. - - >>> src = {'foo': 'bar', 'numbers': [1, 3, 4, 8]} - >>> dst = {'baz': 'qux', 'numbers': [1, 4, 7]} - >>> patch = JsonPatch.from_diff(src, dst) - >>> new = patch.apply(src) - >>> new == dst - True - """ - json_dumper = dumps or cls.json_dumper - builder = DiffBuilder(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. - - :param obj: Document object. - :type obj: dict - - :param in_place: Tweaks the way how patch would be applied - directly to - specified `obj` or to its copy. - :type in_place: bool - - :return: Modified `obj`. - """ - - 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] - return cls(operation, pointer_cls=self.pointer_cls) - - class PatchOperation(object): """A single operation inside a JSON Patch.""" @@ -681,6 +491,201 @@ class CopyOperation(PatchOperation): return obj +class JsonPatch(object): + 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. + + >>> patch = JsonPatch([ + ... {'op': 'add', 'path': '/foo', 'value': 'bar'}, + ... {'op': 'add', 'path': '/baz', 'value': [1, 2, 3]}, + ... {'op': 'remove', 'path': '/baz/1'}, + ... {'op': 'test', 'path': '/baz', 'value': [1, 3]}, + ... {'op': 'replace', 'path': '/baz/0', 'value': 42}, + ... {'op': 'remove', 'path': '/baz/1'}, + ... ]) + >>> doc = {} + >>> result = patch.apply(doc) + >>> expected = {'foo': 'bar', 'baz': [42]} + >>> result == expected + True + + JsonPatch object is iterable, so you can easily access each patch + statement in a loop: + + >>> lpatch = list(patch) + >>> expected = {'op': 'add', 'path': '/foo', 'value': 'bar'} + >>> lpatch[0] == expected + True + >>> lpatch == patch.patch + True + + Also JsonPatch could be converted directly to :class:`bool` if it contains + any operation statements: + + >>> bool(patch) + True + >>> bool(JsonPatch([])) + False + + This behavior is very handy with :func:`make_patch` to write more readable + code: + + >>> old = {'foo': 'bar', 'numbers': [1, 3, 4, 8]} + >>> new = {'baz': 'qux', 'numbers': [1, 4, 7]} + >>> patch = make_patch(old, new) + >>> if patch: + ... # document have changed, do something useful + ... patch.apply(old) #doctest: +ELLIPSIS + {...} + """ + 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. + + :param patch_str: JSON patch as raw string. + :type patch_str: str + + :param loads: A function of one argument that loads a serialized + JSON string. + :type loads: function + + :param pointer_cls: JSON pointer class to use. + :type pointer_cls: Type[JsonPointer] + + :return: :class:`JsonPatch` instance. + """ + 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 + objects. Json patch would be created for `src` argument against `dst` + one. + + :param src: Data source document object. + :type src: dict + + :param dst: Data source document object. + :type dst: dict + + :param dumps: A function of one argument that produces a serialized + JSON string. + :type dumps: function + + :param pointer_cls: JSON pointer class to use. + :type pointer_cls: Type[JsonPointer] + + :return: :class:`JsonPatch` instance. + + >>> src = {'foo': 'bar', 'numbers': [1, 3, 4, 8]} + >>> dst = {'baz': 'qux', 'numbers': [1, 4, 7]} + >>> patch = JsonPatch.from_diff(src, dst) + >>> new = patch.apply(src) + >>> new == dst + True + """ + json_dumper = dumps or cls.json_dumper + builder = DiffBuilder(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. + + :param obj: Document object. + :type obj: dict + + :param in_place: Tweaks the way how patch would be applied - directly to + specified `obj` or to its copy. + :type in_place: bool + + :return: Modified `obj`. + """ + + 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] + return cls(operation, pointer_cls=self.pointer_cls) + + class DiffBuilder(object): def __init__(self, dumps=json.dumps, pointer_cls=JsonPointer): -- cgit v1.2.1