diff options
author | Stefan Kögl <stefan@skoegl.net> | 2013-11-12 10:58:45 -0800 |
---|---|---|
committer | Stefan Kögl <stefan@skoegl.net> | 2013-11-12 10:58:45 -0800 |
commit | 48f9adb8991deb62ba4f327d5e96766c4d0d55e7 (patch) | |
tree | 8b24cfdfe669e3e39e3a3ff8816e876659c904a4 | |
parent | fc4e0b2e21d6bee2b62830d0a1ffc349e7243455 (diff) | |
parent | 0c4ec87ced2ab50364f23b90c67637473a017903 (diff) | |
download | python-json-patch-48f9adb8991deb62ba4f327d5e96766c4d0d55e7.tar.gz |
Merge pull request #19 from kxepal/pylint
Improve code quality
-rw-r--r-- | jsonpatch.py | 141 |
1 files changed, 77 insertions, 64 deletions
diff --git a/jsonpatch.py b/jsonpatch.py index aec583b..c614666 100644 --- a/jsonpatch.py +++ b/jsonpatch.py @@ -30,9 +30,18 @@ # THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. # +""" Apply JSON-Patches (RFC 6902) """ + from __future__ import unicode_literals -""" Apply JSON-Patches (RFC 6902) """ +import collections +import copy +import functools +import inspect +import json +import sys + +from jsonpointer import JsonPointer, JsonPointerException # Will be parsed by setup.py to determine package metadata __author__ = 'Stefan Kögl <stefan@skoegl.net>' @@ -40,21 +49,15 @@ __version__ = '1.3' __website__ = 'https://github.com/stefankoegl/python-json-patch' __license__ = 'Modified BSD License' -import copy -import sys -import operator -import collections - -import json - -import jsonpointer +# pylint: disable=E0611,W0404 if sys.version_info >= (3, 0): - basestring = (bytes, str) + basestring = (bytes, str) # pylint: disable=C0103,W0622 + from itertools import zip_longest +else: + from itertools import izip_longest as zip_longest -JsonPointerException = jsonpointer.JsonPointerException - class JsonPatchException(Exception): """Base Json Patch exception""" @@ -67,6 +70,7 @@ class JsonPatchConflict(JsonPatchException): - etc. """ + class JsonPatchTestFailed(JsonPatchException, AssertionError): """ A Test operation failed """ @@ -74,15 +78,15 @@ class JsonPatchTestFailed(JsonPatchException, AssertionError): def multidict(ordered_pairs): """Convert duplicate keys values to lists.""" # read all values into lists - d = collections.defaultdict(list) - for k, v in ordered_pairs: - d[k].append(v) + mdict = collections.defaultdict(list) + for key, value in ordered_pairs: + mdict[key].append(value) - # unpack lists that have only 1 item - for k, v in d.items(): - if len(v) == 1: - d[k] = v[0] - return dict(d) + return dict( + # unpack lists that have only 1 item + (key, values[0] if len(values) == 1 else values) + for key, values in mdict.items() + ) def get_loadjson(): @@ -94,9 +98,6 @@ def get_loadjson(): function with object_pairs_hook set to multidict for Python versions that support the parameter. """ - import inspect - import functools - argspec = inspect.getargspec(json.load) if 'object_pairs_hook' not in argspec.args: return json.load @@ -123,12 +124,14 @@ def apply_patch(doc, patch, in_place=False): :rtype: dict >>> doc = {'foo': 'bar'} - >>> other = apply_patch(doc, [{'op': 'add', 'path': '/baz', 'value': 'qux'}]) + >>> patch = [{'op': 'add', 'path': '/baz', 'value': 'qux'}] + >>> other = apply_patch(doc, patch) >>> doc is not other True >>> other == {'foo': 'bar', 'baz': 'qux'} True - >>> apply_patch(doc, [{'op': 'add', 'path': '/baz', 'value': 'qux'}], in_place=True) == {'foo': 'bar', 'baz': 'qux'} + >>> patch = [{'op': 'add', 'path': '/baz', 'value': 'qux'}] + >>> apply_patch(doc, patch, in_place=True) == {'foo': 'bar', 'baz': 'qux'} True >>> doc == other True @@ -140,6 +143,7 @@ def apply_patch(doc, patch, in_place=False): patch = JsonPatch(patch) return patch.apply(doc, in_place) + def make_patch(src, dst): """Generates patch by comparing of two document objects. Actually is a proxy to :meth:`JsonPatch.from_diff` method. @@ -230,18 +234,16 @@ class JsonPatch(object): 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 - return len(list(self._ops)) == len(list(other._ops)) and \ - all(map(operator.eq, self._ops, other._ops)) - + def __ne__(self, other): + return not(self == other) @classmethod def from_string(cls, patch_str): @@ -298,7 +300,9 @@ class JsonPatch(object): yield operation for key in dst: if key not in src: - yield {'op': 'add', 'path': '/'.join(path + [key]), 'value': dst[key]} + yield {'op': 'add', + 'path': '/'.join(path + [key]), + 'value': dst[key]} def compare_list(path, src, dst): lsrc, ldst = len(src), len(dst) @@ -309,7 +313,9 @@ class JsonPatch(object): if lsrc < ldst: for idx in range(lsrc, ldst): current = path + [str(idx)] - yield {'op': 'add', 'path': '/'.join(current), 'value': dst[idx]} + yield {'op': 'add', + 'path': '/'.join(current), + 'value': dst[idx]} elif lsrc > ldst: for idx in reversed(range(ldst, lsrc)): yield {'op': 'remove', 'path': '/'.join(path + [str(idx)])} @@ -322,7 +328,7 @@ class JsonPatch(object): @property def _ops(self): - return map(self._get_operation, self.patch) + return tuple(map(self._get_operation, self.patch)) def apply(self, obj, in_place=False): """Applies the patch to given object. @@ -355,36 +361,35 @@ class JsonPatch(object): raise JsonPatchException("Operation must be a string") if op not in self.operations: - raise JsonPatchException("Unknown operation '%s'" % op) + raise JsonPatchException("Unknown operation {0!r}".format(op)) cls = self.operations[op] return cls(operation) - class PatchOperation(object): """A single operation inside a JSON Patch.""" def __init__(self, operation): self.location = operation['path'] - self.pointer = jsonpointer.JsonPointer(self.location) + self.pointer = JsonPointer(self.location) self.operation = operation def apply(self, obj): """Abstract method that applies patch operation to specified object.""" raise NotImplementedError('should implement 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) + class RemoveOperation(PatchOperation): """Removes an object property or an array element.""" @@ -406,30 +411,25 @@ class AddOperation(PatchOperation): value = self.operation["value"] subobj, part = self.pointer.to_last(obj) - # type is already checked in to_last(), so we assert here - # for consistency - assert isinstance(subobj, list) or isinstance(subobj, dict), \ - "invalid document type %s" (type(doc),) - if isinstance(subobj, list): - if part == '-': - subobj.append(value) + 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) + subobj.insert(part, value) # pylint: disable=E1103 elif isinstance(subobj, dict): if part is None: - # we're replacing the root - obj = value - + obj = value # we're replacing the root else: subobj[part] = value + else: + raise TypeError("invalid document type {0}".format(type(subobj))) + return obj @@ -440,11 +440,6 @@ class ReplaceOperation(PatchOperation): value = self.operation["value"] subobj, part = self.pointer.to_last(obj) - # type is already checked in to_last(), so we assert here - # for consistency - assert isinstance(subobj, list) or isinstance(subobj, dict), \ - "invalid document type %s" (type(doc),) - if part is None: return value @@ -454,8 +449,10 @@ class ReplaceOperation(PatchOperation): elif isinstance(subobj, dict): if not part in subobj: - raise JsonPatchConflict("can't replace non-existant object '%s'" - "" % part) + 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 @@ -465,15 +462,24 @@ class MoveOperation(PatchOperation): """Moves an object property or an array element to new location.""" def apply(self, obj): - from_ptr = jsonpointer.JsonPointer(self.operation['from']) + from_ptr = JsonPointer(self.operation['from']) subobj, part = from_ptr.to_last(obj) value = subobj[part] if self.pointer.contains(from_ptr): raise JsonPatchException('Cannot move values into its own children') - obj = RemoveOperation({'op': 'remove', 'path': self.operation['from']}).apply(obj) - obj = AddOperation({'op': 'add', 'path': self.location, 'value': value}).apply(obj) + obj = RemoveOperation({ + 'op': 'remove', + 'path': self.operation['from'] + }).apply(obj) + + obj = AddOperation({ + 'op': 'add', + 'path': self.location, + 'value': value + }).apply(obj) + return obj @@ -487,14 +493,15 @@ class TestOperation(PatchOperation): val = subobj else: val = self.pointer.walk(subobj, part) - except JsonPointerException as ex: raise JsonPatchTestFailed(str(ex)) if 'value' in self.operation: value = self.operation['value'] if val != value: - raise JsonPatchTestFailed('%s is not equal to tested value %s (types %s and %s)' % (val, value, type(val), type(value))) + msg = '{0} ({1}) is not equal to tested value {2} ({3})' + raise JsonPatchTestFailed(msg.format(val, type(val), + value, type(value))) return obj @@ -503,8 +510,14 @@ class CopyOperation(PatchOperation): """ Copies an object property or an array element to a new location """ def apply(self, obj): - from_ptr = jsonpointer.JsonPointer(self.operation['from']) + from_ptr = JsonPointer(self.operation['from']) subobj, part = from_ptr.to_last(obj) value = copy.deepcopy(subobj[part]) - obj = AddOperation({'op': 'add', 'path': self.location, 'value': value}).apply(obj) + + obj = AddOperation({ + 'op': 'add', + 'path': self.location, + 'value': value + }).apply(obj) + return obj |