add path tracking

Author: Korijn

.gitignore

@@ -1,3 +1,5 @@
-*.pyc
+*.egg-info/
+__pycache__
 .vscode
-dist
+.coverage
+dist

patchdiff/__init__.py

@@ -1,78 +1,3 @@
-from typing import Dict, List
-
+from .diff import diff
 
 __version__ = "0.2.0"
-
-
-def diff_lists(input: List, output: List):
-    memory = {(0, 0): {"ops": [], "cost": 0}}
-
-    def dist(i, j):
-        if (i, j) not in memory:
-            if i > 0 and j > 0 and not diff(input[i - 1], output[j - 1]):
-                step = dist(i - 1, j - 1)
-            else:
-                paths = []
-                if i > 0:
-                    base = dist(i - 1, j)
-                    op = {"op": "remove", "idx": i - 1}
-                    paths.append(
-                        {
-                            "ops": base["ops"] + [op],
-                            "cost": base["cost"] + 1,
-                        }
-                    )
-                if j > 0:
-                    base = dist(i, j - 1)
-                    op = {"op": "add", "idx": j - 1, "value": output[j - 1]}
-                    paths.append(
-                        {
-                            "ops": base["ops"] + [op],
-                            "cost": base["cost"] + 1,
-                        }
-                    )
-                if i > 0 and j > 0:
-                    base = dist(i - 1, j - 1)
-                    op = {
-                        "op": "replace",
-                        "idx": i - 1,
-                        "original": input[i - 1],
-                        "value": output[j - 1],
-                    }
-                    paths.append(
-                        {
-                            "ops": base["ops"] + [op],
-                            "cost": base["cost"] + 1,
-                        }
-                    )
-                step = min(paths, key=lambda a: a["cost"])
-            memory[(i, j)] = step
-        return memory[(i, j)]
-
-    return dist(len(input), len(output))["ops"]
-
-
-def diff_dicts(input: Dict, output: Dict):
-    ops = []
-    input_keys = set(input.keys())
-    output_keys = set(output.keys())
-    for key in input_keys - output_keys:
-        ops.append({"op": "remove", "key": key})
-    for key in output_keys - input_keys:
-        ops.append({"op": "add", "key": key, "value": output[key]})
-    for key in input_keys & output_keys:
-        ops.extend(diff(input[key], output[key]))
-    return ops
-
-
-def diff(input, output):
-    # TODO: track paths
-    # TODO: properly check equality
-    if input == output:
-        return []
-    if isinstance(input, list) and isinstance(output, list):
-        return diff_lists(input, output)
-    if isinstance(input, dict) and isinstance(output, dict):
-        return diff_dicts(input, output)
-    # TODO: sets, tuples
-    return [{"op": "replace", "value": output}]

patchdiff/diff.py

@@ -0,0 +1,123 @@
+from functools import reduce
+from typing import Dict, List, Set
+
+from .pointer import Pointer
+from .types import Diffable
+
+
+def diff_lists(input: List, output: List, ptr: Pointer) -> List:
+    memory = {(0, 0): {"ops": [], "cost": 0}}
+
+    def dist(i, j):
+        if (i, j) not in memory:
+            if i > 0 and j > 0 and input[i - 1] == output[j - 1]:
+                step = dist(i - 1, j - 1)
+            else:
+                paths = []
+                if i > 0:
+                    base = dist(i - 1, j)
+                    op = {"op": "remove", "idx": i - 1}
+                    paths.append(
+                        {
+                            "ops": base["ops"] + [op],
+                            "cost": base["cost"] + 1,
+                        }
+                    )
+                if j > 0:
+                    base = dist(i, j - 1)
+                    op = {"op": "add", "idx": j - 1, "value": output[j - 1]}
+                    paths.append(
+                        {
+                            "ops": base["ops"] + [op],
+                            "cost": base["cost"] + 1,
+                        }
+                    )
+                if i > 0 and j > 0:
+                    base = dist(i - 1, j - 1)
+                    op = {
+                        "op": "replace",
+                        "idx": i - 1,
+                        "original": input[i - 1],
+                        "value": output[j - 1],
+                    }
+                    paths.append(
+                        {
+                            "ops": base["ops"] + [op],
+                            "cost": base["cost"] + 1,
+                        }
+                    )
+                step = min(paths, key=lambda a: a["cost"])
+            memory[(i, j)] = step
+        return memory[(i, j)]
+
+    ops = dist(len(input), len(output))["ops"]
+
+    def pad(state, op):
+        ops, padding = state
+        if op["op"] == "add":
+            padded_idx = op["idx"] + 1 + padding
+            idx_token = str(padded_idx) if padded_idx < len(input) + padding else "-"
+            full_op = {
+                "op": "add",
+                "path": str(ptr.append(idx_token)),
+                "value": op["value"],
+            }
+            return [ops + [full_op], padding + 1]
+        elif op["op"] == "remove":
+            full_op = {
+                "op": "remove",
+                "path": str(ptr.append(str(op["idx"] + padding))),
+            }
+            return [ops + [full_op], padding - 1]
+        else:
+            replace_ptr = ptr.append(str(op["idx"] + padding))
+            replace_ops = diff(op["original"], op["value"], replace_ptr)
+            return [ops + replace_ops, padding]
+
+    padded_ops, _ = reduce(pad, ops, [[], 0])
+
+    return padded_ops
+
+
+def diff_dicts(input: Dict, output: Dict, ptr: Pointer) -> List:
+    ops = []
+    input_keys = set(input.keys())
+    output_keys = set(output.keys())
+    for key in input_keys - output_keys:
+        ops.append({"op": "remove", "path": str(ptr.append(key)), "key": key})
+    for key in output_keys - input_keys:
+        ops.append(
+            {
+                "op": "add",
+                "path": str(ptr.append(key)),
+                "key": key,
+                "value": output[key],
+            }
+        )
+    for key in input_keys & output_keys:
+        ops.extend(diff(input[key], output[key], ptr.append(key)))
+    return ops
+
+
+def diff_sets(input: Set, output: Set, ptr: Pointer) -> List:
+    # TODO: pointers?
+    ops = []
+    for value in input - output:
+        ops.append({"op": "remove", "value": value})
+    for value in output - input:
+        ops.append({"op": "add", "value": value})
+    return ops
+
+
+def diff(input: Diffable, output: Diffable, ptr: Pointer = None) -> List:
+    if input == output:
+        return []
+    if ptr is None:
+        ptr = Pointer()
+    if isinstance(input, list) and isinstance(output, list):
+        return diff_lists(input, output, ptr)
+    if isinstance(input, dict) and isinstance(output, dict):
+        return diff_dicts(input, output, ptr)
+    if isinstance(input, set) and isinstance(output, set):
+        return diff_sets(input, output, ptr)
+    return [{"op": "replace", "path": str(ptr), "value": output}]

patchdiff/pointer.py

@@ -0,0 +1,62 @@
+import re
+from typing import List
+
+from .types import Diffable
+
+
+tilde0_re = re.compile("~0")
+tilde1_re = re.compile("~1")
+tilde_re = re.compile("~")
+slash_re = re.compile("/")
+
+
+def unescape(token: str) -> str:
+    return tilde0_re.sub("~", tilde1_re.sub("/", token))
+
+
+def escape(token: str) -> str:
+    return slash_re.sub("~1", tilde_re.sub("~0", token))
+
+
+class Pointer:
+    def __init__(self, tokens: List[str] = None) -> None:
+        if tokens is None:
+            tokens = [""]
+        self.tokens = tokens
+
+    @staticmethod
+    def from_json(path: str) -> "Pointer":
+        tokens = [unescape(t) for t in path.split("/")]
+        return Pointer(tokens)
+
+    def __str__(self) -> str:
+        return "/".join(escape(t) for t in self.tokens)
+
+    def evaluate(self, obj: Diffable):
+        key = ""
+        parent = None
+        value = obj
+        for key in self.tokens[1:]:
+            parent = value
+            if isinstance(parent, list):
+                key = int(key)
+            value = parent[key]
+        return parent, key, value
+
+    def get(self, obj: Diffable):
+        _, _, value = self.evaluate(obj)
+        return value
+
+    def set(self, obj: Diffable, value):
+        cursor = obj
+        for key in self.tokens[1:-1]:
+            cursor = cursor[key]
+        cursor[self.tokens[-1]] = value
+
+    def iappend(self, token):
+        """append, in-place"""
+        self.tokens.append(token)
+
+    def append(self, token):
+        """append, creating new Pointer"""
+        return Pointer(self.tokens + [token])

patchdiff/types.py

@@ -0,0 +1,3 @@
+from typing import Dict, List, Set, Tuple, Union
+
+Diffable = Union[Dict, List, Set, Tuple]

setup.cfg

@@ -5,8 +5,8 @@ max-line-length = 88
 extend-ignore =
     # See https://github.com/PyCQA/pycodestyle/issues/373
     E203,
-application-import-names = observ
+application-import-names = patchdiff
 import-order-style = google
 per-file-ignores =
-    observ/__init__.py:F401,F403
+    patchdiff/__init__.py:F401,F403
 exclude = .venv

tests/test_all.py renamed to tests/test_diff.py

@@ -7,9 +7,9 @@ def test_list():
     ops = diff(a, b)
 
     assert len(ops) == 3
-    assert (ops[0]["op"], ops[0]["idx"]) == ("replace", 0)
-    assert (ops[1]["op"], ops[1]["idx"]) == ("replace", 3)
-    assert (ops[2]["op"], ops[2]["idx"]) == ("remove", 4)
+    assert (ops[0]["op"], ops[0]["path"]) == ("replace", "/0")
+    assert (ops[1]["op"], ops[1]["path"]) == ("replace", "/3")
+    assert (ops[2]["op"], ops[2]["path"]) == ("remove", "/4")
 
 
 def test_dicts():

tests/test_pointer.py

@@ -0,0 +1,19 @@
+from patchdiff.pointer import Pointer
+
+
+def test_pointer_get():
+    obj = [1, 5, {"foo": 1, "bar": [1, 2, 3]}, "sdfsdf", "fff"]
+    assert Pointer(["", "1"]).get(obj) == 5
+    assert Pointer(["", "2", "bar", "1"]).get(obj) == 2
+
+
+def test_pointer_str():
+    assert str(Pointer(["", "1"])) == "/1"
+    assert str(Pointer(["", "foo", "bar", "-"])) == "/foo/bar/-"
+
+
+def test_pointer_iappend():
+    ptr = Pointer()
+    ptr.iappend("3")
+    ptr.iappend("foo")
+    assert str(ptr) == "/3/foo"