Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions README.md
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,7 +4,7 @@ Diff is a library to calculate deltas between structured data.

## Features

- Calculate detla(s)
- Calculate delta(s)
- Rebuild state from delta(s)

## Supported Formats
Expand DownExpand Up@@ -47,7 +47,7 @@ deltas = diff.diff(new=new, old=old)

rebuild_new = diff.patch(base=old, deltas=deltas)

assert rebuild_new == old
assert rebuild_new == new
```

## Install
Expand Down
726 changes: 483 additions & 243 deletions poetry.lock

Large diffs are not rendered by default.

4 changes: 2 additions & 2 deletions src/diff/__init__.py
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
from diff.delta import Delta
from diff.delta import Delta, JsonValue
from diff.diff import diff
from diff.patch import patch

__all__ = ["Delta", "diff", "patch"]
__all__ = ["Delta", "JsonValue", "diff", "patch"]
7 changes: 5 additions & 2 deletions src/diff/delta.py
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,16 @@
import dataclasses
import typing

JsonScalar: typing.TypeAlias = str | int | float | bool | None
JsonValue: typing.TypeAlias = JsonScalar | list["JsonValue"] | dict[str, "JsonValue"]


@dataclasses.dataclass
class Delta:
operation: typing.Literal["deleted", "modified", "added"]
path: str
new_value: typing.Any | None
old_value: typing.Any | None
new_value: JsonValue
old_value: JsonValue

def __repr__(self):
return (
Expand Down
35 changes: 26 additions & 9 deletions src/diff/diff.py
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,37 @@
import typing

from diff import json_path
from diff.delta import Delta
from diff.delta import Delta, JsonValue


def _path_sort_key(path: str) -> tuple[tuple[int, str | int], ...]:
return tuple(
(0, token) if isinstance(token, str) else (1, token)
for token in json_path.tokenize_json_path(path)
)


def diff(new: JsonValue, old: JsonValue) -> list[Delta]:
if new == old:
return []

if (
not isinstance(new, (dict, list))
or not isinstance(old, (dict, list))
or type(new) is not type(old)
):
return [Delta(path="$", operation="modified", old_value=old, new_value=new)]

def diff(new: dict[str, typing.Any], old: dict[str, typing.Any]) -> list[Delta]:
new_path_map = json_path.path_value_map(
new, include_root=True, leaves_only=True, include_containers=False
new, include_root=True, leaves_only=True, include_containers=True
)
old_path_map = json_path.path_value_map(
old, include_root=True, leaves_only=False, include_containers=False
old, include_root=True, leaves_only=True, include_containers=True
)
new_path_map.pop("$", None)
old_path_map.pop("$", None)
operations: list[Delta] = []

deleted = old_path_map.keys() - new_path_map.keys()
for key in deleted:
for key in sorted(deleted, key=_path_sort_key, reverse=True):
operations.append( # noqa: PERF401
Delta(
path=key,
Expand All@@ -25,15 +42,15 @@ def diff(new: dict[str, typing.Any], old: dict[str, typing.Any]) -> list[Delta]:
)

added = new_path_map.keys() - old_path_map.keys()
for key in added:
for key in sorted(added, key=_path_sort_key):
operations.append( # noqa: PERF401
Delta(
path=key, operation="added", old_value=None, new_value=new_path_map[key]
)
)

shared_keys = new_path_map.keys() & old_path_map.keys()
for key in shared_keys:
for key in sorted(shared_keys, key=_path_sort_key):
if old_path_map[key] != new_path_map[key]:
operations.append( # noqa: PERF401
Delta(
Expand Down
84 changes: 84 additions & 0 deletions src/diff/json_path.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -4,6 +4,90 @@
Token = str | int


class JsonPathError(ValueError):
pass


def tokenize_json_path(path: str) -> list[Token]:
"""Tokenize a JSONPath-like path into dictionary keys and list indices."""
if not isinstance(path, str) or not path:
raise JsonPathError("Path must be a non-empty string.")

i = 0
n = len(path)
tokens: list[Token] = []

if path.startswith("$"):
i += 1
if i < n and path[i] == ".":
i += 1

def read_simple_key(start: int) -> tuple[str, int]:
j = start
while j < n and path[j] not in ".[":
j += 1
if j == start:
raise JsonPathError(f"Expected key at position {start} in '{path}'")
return path[start:j], j

def read_bracket_key_or_index(start: int) -> tuple[Token, int]:
j = start + 1
if j >= n:
raise JsonPathError(f"Unclosed '[' at position {start} in '{path}'")

if path[j] in ("'", '"'):
quote = path[j]
j += 1
buffer: list[str] = []
while j < n:
char = path[j]
if char == "\\":
j += 1
if j >= n:
raise JsonPathError("Trailing backslash in quoted key.")
buffer.append(path[j])
j += 1
continue
if char == quote:
j += 1
break
buffer.append(char)
j += 1
else:
raise JsonPathError(
f"Unclosed quoted key starting at position {start} in '{path}'"
)

if j >= n or path[j] != "]":
raise JsonPathError(
f"Expected ']' after quoted key at position {j} in '{path}'"
)
return "".join(buffer), j + 1

k = j
while k < n and path[k].isdigit():
k += 1
if k == j:
raise JsonPathError(
f"Expected non-negative integer index after '[' at position {start} in '{path}'"
)
if k >= n or path[k] != "]":
raise JsonPathError(f"Expected ']' after index at position {k} in '{path}'")
return int(path[j:k]), k + 1

while i < n:
if path[i] == ".":
i += 1
elif path[i] == "[":
token, i = read_bracket_key_or_index(i)
tokens.append(token)
else:
token, i = read_simple_key(i)
tokens.append(token)

return tokens


def _escape_key_for_brackets(key: str) -> str:
"""Escape a key for bracket notation with double quotes."""
return key.replace("\\", "\\\\").replace('"', '\\"')
Expand Down
127 changes: 11 additions & 116 deletions src/diff/patch.py
Original file line numberDiff line numberDiff line change
@@ -1,118 +1,8 @@
import copy
import typing
from typing import Any

from diff.diff import Delta

Token = str | int # str for dict keys, int for list indices


class JsonPathError(ValueError):
pass


def _tokenize_json_path(path: str) -> list[Token]:
"""
Tokenize a JSONPath-like string into a list of tokens.
- Dict keys -> strings
- List indices -> integers
Supported syntax:
$.a.b[2]["key.with.dots"][0]
Notes:
- Leading '$' is optional.
- Dot-notation for simple keys.
- Brackets for indices and quoted keys. Quotes can be ' or ".
- Escaping inside quoted keys: backslash escapes the quote and backslash (\", \', \\).
"""
if not isinstance(path, str) or not path:
raise JsonPathError("Path must be a non-empty string.")

i = 0
n = len(path)
tokens: list[Token] = []

# Skip optional leading '$' and optional following '.'
if i < n and path[i] == "$":
i += 1
if i < n and path[i] == ".":
i += 1

def read_simple_key(start: int) -> tuple[str, int]:
j = start
while j < n and path[j] not in ".[":
j += 1
if j == start:
raise JsonPathError(f"Expected key at position {start} in '{path}'")
return path[start:j], j

def read_bracket_key_or_index(start: int) -> tuple[Token, int]:
# start at '[', return (token, new_index_after_'])
j = start + 1
if j >= n:
raise JsonPathError(f"Unclosed '[' at position {start} in '{path}'")

if path[j] in ("'", '"'):
# Quoted key
quote = path[j]
j += 1
buf = []
while j < n:
ch = path[j]
if ch == "\\": # escape sequence
j += 1
if j >= n:
raise JsonPathError("Trailing backslash in quoted key.")
esc = path[j]
if esc in [quote, "\\"]:
buf.append(esc)
else:
# Keep unknown escape as-is (e.g., \n), or handle specially if desired
buf.append(esc)
j += 1
continue
if ch == quote:
j += 1
break
buf.append(ch)
j += 1
else:
raise JsonPathError(
f"Unclosed quoted key starting at position {start} in '{path}'"
)

# Expect closing ']'
if j >= n or path[j] != "]":
raise JsonPathError(
f"Expected ']' after quoted key at position {j} in '{path}'"
)
return "".join(buf), j + 1

# Numeric index
k = j
while k < n and path[k].isdigit():
k += 1
if k == j:
raise JsonPathError(
f"Expected non-negative integer index after '[' at position {start} in '{path}'"
)
if k >= n or path[k] != "]":
raise JsonPathError(f"Expected ']' after index at position {k} in '{path}'")
idx = int(path[j:k])
return idx, k + 1

while i < n:
ch = path[i]
if ch == ".":
i += 1 # skip redundant dots (e.g., '$.a..b' would error on next step)
continue
elif ch == "[":
token, i = read_bracket_key_or_index(i)
tokens.append(token)
else:
key, i = read_simple_key(i)
tokens.append(key)

return tokens
from diff.delta import Delta, JsonValue
from diff.json_path import JsonPathError, Token, tokenize_json_path


def set_by_json_path(
Expand All@@ -129,7 +19,7 @@ def set_by_json_path(
- TypeError when the path expects a dict/list but finds another type.
- IndexError for negative indices (not supported).
"""
tokens = _tokenize_json_path(path)
tokens = tokenize_json_path(path)
if not tokens:
raise JsonPathError("Path resolves to the root; set on '$' is not supported.")

Expand DownExpand Up@@ -199,7 +89,7 @@ def get_by_json_path(doc: Any, path: str) -> Any:
"""
Retrieve a value from `doc` following the same JSONPath-like syntax.
"""
tokens = _tokenize_json_path(path)
tokens = tokenize_json_path(path)
current = doc
for idx, tok in enumerate(tokens):
if isinstance(tok, str):
Expand DownExpand Up@@ -262,7 +152,7 @@ def pop_by_json_path(
Remove and return the value at `path`. Behaves like delete_by_json_path but returns the removed value.
If the path is missing and missing_ok=True, returns None and leaves `doc` unchanged.
"""
tokens = _tokenize_json_path(path)
tokens = tokenize_json_path(path)
if not tokens:
raise JsonPathError("Path resolves to the root; popping '$' is not supported.")

Expand DownExpand Up@@ -353,9 +243,14 @@ def pop_by_json_path(
return removed


def patch(base: dict[str, typing.Any], deltas: list[Delta]) -> dict[str, typing.Any]:
def patch(base: JsonValue, deltas: list[Delta]) -> JsonValue:
output = copy.deepcopy(base)
for op in deltas:
if op.path == "$":
if op.operation == "deleted":
raise JsonPathError("Deleting the root value is not supported.")
output = copy.deepcopy(op.new_value)
continue
if op.operation == "deleted":
pop_by_json_path(output, op.path, prune_empty=True, remove_from_list=True)
continue
Expand Down
Loading
Loading