#!/usr/bin/env python3 """jsonptr_patch.py -- RFC 6901 JSON Pointer + RFC 6902 JSON Patch. Pure stdlib (json, copy, sys). No third-party imports. Public API ---------- JSON Pointer (RFC 6901 s.3/s.4): decode_pointer(ref) -> tuple[str, ...] validate + unescape unescape_token(tok) -> str '~1' -> '/', '~0' -> '~' escape_token(tok) -> str inverse pointer_to_string(toks) -> str inverse of decode_pointer resolve(doc, tokens) -> value strict read, raises on miss JSON Patch (RFC 6902 s.4): json_equal(a, b) -> bool JSON value equality apply_patch(doc, patch) -> new document input is NEVER mutated Errors: JsonPointerError and JsonPatchError (both ValueError subclasses). Semantics, measured against the vendored official suite json-patch/json-patch-tests (tests.json + spec_tests.json, sha256 pinned in test_jsonptr_patch.py): * `add` never creates missing intermediate objects -- add /a/b against {} is an error (spec_tests.json case "A.12. Adding to a Non-existent Target"). * array `add`: index == len(array) is legal, index > len(array) is an error, "-" appends. * `remove` / `replace` / `copy` / `test` require the target to exist; `move` requires `from` to exist and forbids moving into a strict descendant of `from`. * `test` uses JSON value equality: 1 == 1.0, true != 1, null == null, object member order is irrelevant, array element order is significant. * the patch is atomic: every op is applied to a deep copy and the copy is returned only if all ops succeed; the caller's document is untouched either way. CLI: python3 jsonptr_patch.py --selftest [--verbose] python3 jsonptr_patch.py --selftest-json python3 jsonptr_patch.py --doc doc.json --patch patch.json Exit codes: 0 = success, 1 = RFC 6901/6902 error, 2 = usage error. """ from __future__ import annotations import copy import json import sys __all__ = [ "JsonPointerError", "JsonPatchError", "decode_pointer", "unescape_token", "escape_token", "pointer_to_string", "resolve", "json_equal", "apply_patch", "run_selftest", ] PATCH_ERR = 1 USAGE_ERR = 2 class JsonPointerError(ValueError): """Malformed JSON Pointer string (RFC 6901 s.3/s.4).""" class JsonPatchError(ValueError): """Malformed patch, or a failing operation (RFC 6902 s.4).""" # --------------------------------------------------------------------------- # RFC 6901 -- JSON Pointer # --------------------------------------------------------------------------- def escape_token(tok: str) -> str: """'~' -> '~0' first, then '/' -> '~1' (order matters: '~/' -> '~0~1').""" return str(tok).replace("~", "~0").replace("/", "~1") def unescape_token(tok: str) -> str: """Inverse of escape_token. '~0' -> '~', '~1' -> '/'; anything else is illegal.""" out = [] i, n = 0, len(tok) while i < n: ch = tok[i] if ch != "~": out.append(ch) i += 1 continue if i + 1 >= n: raise JsonPointerError(f"dangling '~' at end of reference token {tok!r}") nxt = tok[i + 1] if nxt == "0": out.append("~") elif nxt == "1": out.append("/") else: raise JsonPointerError( f"invalid escape '~{nxt}' in reference token {tok!r} " "(only ~0 and ~1 are legal)" ) i += 2 return "".join(out) def decode_pointer(ref: str) -> tuple[str, ...]: """Split a JSON Pointer string into unescaped reference tokens. Grammar (RFC 6901 s.3): json-pointer = *( "/" reference-token ) "" -> () the whole document "/a/b" -> ("a", "b") "a" -> rejected: must be empty or start with "/" "/a~2b" -> rejected: '~2' is not a legal escape """ if not isinstance(ref, str): raise JsonPointerError(f"pointer must be a str, got {type(ref).__name__}") if ref == "": return () if not ref.startswith("/"): raise JsonPointerError(f"pointer {ref!r} must be empty or start with '/'") return tuple(unescape_token(t) for t in ref[1:].split("/")) def pointer_to_string(tokens) -> str: """Inverse of decode_pointer: () -> "", ("a", "b") -> "/a/b".""" return "".join("/" + escape_token(t) for t in tokens) def is_array_index(tok: str) -> bool: """RFC 6901 s.4: digits only, and a leading zero is only allowed for '0'. ``str.isdigit()`` is NOT sufficient: it is True for Unicode digit glyphs such as ``"²"`` (SUPERSCRIPT TWO) and ``"١"`` (ARABIC-INDIC ONE), whose ``int()`` raises a plain ``ValueError`` -- which would leak past the JsonPointerError contract. ``str.isascii()`` pins the check to the ASCII digits that ``int()`` actually accepts. """ if not tok or not tok.isascii() or not tok.isdigit(): return False return tok == "0" or tok[0] != "0" def _descend(node, tok: str): """Walk one token. Arrays accept an RFC 6901 array-index token.""" if isinstance(node, dict): if tok not in node: raise JsonPointerError(f"key {tok!r} not found in object") return node[tok] if isinstance(node, list): if not is_array_index(tok): raise JsonPointerError(f"invalid array index {tok!r}") idx = int(tok) if idx >= len(node): raise JsonPointerError( f"array index {tok!r} out of range (len={len(node)})" ) return node[idx] raise JsonPointerError( f"cannot index {type(node).__name__} with {tok!r}: not an object or array" ) def resolve(doc, tokens) -> object: """Evaluate a token tuple against `doc`; raise JsonPointerError on any miss.""" node = doc for tok in tokens: node = _descend(node, tok) return node # --------------------------------------------------------------------------- # JSON value equality (used by the `test` operation) # --------------------------------------------------------------------------- def json_equal(a, b) -> bool: """RFC 6902 `test` semantics. Numbers compare numerically (1 == 1.0); booleans are a distinct type from numbers (True != 1); null equals only null; object member order is irrelevant; array order is significant. """ if isinstance(a, bool) or isinstance(b, bool): return isinstance(a, bool) and isinstance(b, bool) and a is b if a is None or b is None: return a is None and b is None if isinstance(a, (int, float)) and isinstance(b, (int, float)): return a == b if isinstance(a, str) or isinstance(b, str): return isinstance(a, str) and isinstance(b, str) and a == b if isinstance(a, dict): if not isinstance(b, dict) or set(a) != set(b): return False return all(json_equal(a[k], b[k]) for k in a) if isinstance(b, dict): return False if isinstance(a, list): if not isinstance(b, list) or len(a) != len(b): return False return all(json_equal(x, y) for x, y in zip(a, b)) return False # --------------------------------------------------------------------------- # RFC 6902 -- JSON Patch # --------------------------------------------------------------------------- _OPS = ("add", "remove", "replace", "move", "copy", "test") def _canonical(value) -> str: """Canonical JSON, used only for error messages.""" return json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":")) def _require(cond: bool, msg: str) -> None: if not cond: raise JsonPatchError(msg) def _parent_and_token(doc, tokens): """Return (container, final_token) for a non-root pointer, validated.""" _require(bool(tokens), "operation target must not be the whole document") container = resolve(doc, tokens[:-1]) tok = tokens[-1] if isinstance(container, dict): _require(tok in container, f"path {pointer_to_string(tokens)!r} does not exist") elif isinstance(container, list): _require(tok != "-", "'-' is not a valid target for this operation") _require(is_array_index(tok), f"invalid array index {tok!r}") _require(int(tok) < len(container), f"array index {tok!r} out of range (len={len(container)})") else: raise JsonPatchError( f"path {pointer_to_string(tokens)!r} targets a scalar" ) return container, tok def _op_add(doc, tokens, value): if not tokens: # root replacement return copy.deepcopy(value) container = resolve(doc, tokens[:-1]) # strict: parents must exist tok = tokens[-1] if isinstance(container, list): if tok == "-": container.append(copy.deepcopy(value)) return doc _require(is_array_index(tok), f"invalid array index {tok!r}") idx = int(tok) _require(idx <= len(container), f"array index {tok!r} is greater than number of items " f"in array (len={len(container)})") container.insert(idx, copy.deepcopy(value)) return doc if isinstance(container, dict): container[tok] = copy.deepcopy(value) # create-or-replace return doc raise JsonPatchError( f"cannot add into {type(container).__name__} at " f"{pointer_to_string(tokens)!r}" ) def _op_remove(doc, tokens): container, tok = _parent_and_token(doc, tokens) if isinstance(container, dict): del container[tok] else: del container[int(tok)] return doc def _op_replace(doc, tokens, value): if not tokens: return copy.deepcopy(value) container, tok = _parent_and_token(doc, tokens) if isinstance(container, dict): container[tok] = copy.deepcopy(value) else: container[int(tok)] = copy.deepcopy(value) return doc def _op_move(doc, tokens, from_tokens): _require(bool(from_tokens), "'move' 'from' must not be the whole document") if pointer_to_string(from_tokens) == pointer_to_string(tokens): return doc # /a -> /a is a no-op # Moving an object into its own descendant would destroy the target path. _require( not (len(tokens) > len(from_tokens) and tuple(tokens[:len(from_tokens)]) == from_tokens), f"cannot move {pointer_to_string(from_tokens)!r} into its own " f"descendant {pointer_to_string(tokens)!r}", ) moved = copy.deepcopy(resolve(doc, from_tokens)) _op_remove(doc, from_tokens) _op_add(doc, tokens, moved) # `to` resolves on the post-removal doc return doc def _op_copy(doc, tokens, from_tokens): _require(bool(from_tokens), "'copy' 'from' must not be the whole document") src = resolve(doc, from_tokens) _op_add(doc, tokens, copy.deepcopy(src)) return doc def _op_test(doc, tokens, value): actual = resolve(doc, tokens) _require( json_equal(actual, value), f"test failed at {pointer_to_string(tokens)!r}: " f"expected {_canonical(value)!r}, got {_canonical(actual)!r}", ) return doc def _apply(work, op): if not isinstance(op, dict): raise JsonPatchError( f"each patch entry must be an object, got {type(op).__name__}" ) name = op.get("op") _require(isinstance(name, str) and name in _OPS, f"unrecognized op {name!r}") _require("path" in op, f"missing 'path' parameter to {name!r}") if name in ("add", "replace", "test"): _require("value" in op, f"missing 'value' parameter to {name!r}") if name in ("move", "copy"): _require("from" in op, f"missing 'from' parameter to {name!r}") try: tokens = decode_pointer(op["path"]) if name in ("move", "copy"): from_tokens = decode_pointer(op["from"]) else: from_tokens = () except JsonPointerError as exc: raise JsonPatchError(str(exc)) from exc if name == "add": return _op_add(work, tokens, op["value"]) if name == "remove": return _op_remove(work, tokens) if name == "replace": return _op_replace(work, tokens, op["value"]) if name == "move": return _op_move(work, tokens, from_tokens) if name == "copy": return _op_copy(work, tokens, from_tokens) return _op_test(work, tokens, op["value"]) def apply_patch(doc, patch): """Apply an RFC 6902 patch and return the patched document. `doc` is never mutated: the operations run on a deep copy which is returned only when every op succeeds, so a failing patch is atomic (nothing applied). """ if not isinstance(patch, list): raise JsonPatchError("a JSON Patch must be an array of operations") work = copy.deepcopy(doc) for i, op in enumerate(patch): try: work = _apply(work, op) except (JsonPatchError, JsonPointerError) as exc: raise JsonPatchError( f"patch[{i}] ({op.get('op') if isinstance(op, dict) else op!r}): {exc}" ) from exc return work def apply_patch_text(doc_str: str, patch_str: str) -> str: """Parse JSON text, apply the patch, return the result as JSON text.""" result = apply_patch(json.loads(doc_str), json.loads(patch_str)) return json.dumps(result, ensure_ascii=False) # --------------------------------------------------------------------------- # Self-test # --------------------------------------------------------------------------- def _selftest_cases(): """(name, doc, patch, expected_or_None, must_raise).""" return [ # ---- RFC 6902 Appendix A, verbatim ---- ("A.1 add object member", {"foo": "bar"}, [{"op": "add", "path": "/baz", "value": "qux"}], {"foo": "bar", "baz": "qux"}, False), ("A.2 add array element", {"foo": ["bar", "baz"]}, [{"op": "add", "path": "/foo/1", "value": "qux"}], {"foo": ["bar", "qux", "baz"]}, False), ("A.3 remove object member", {"baz": "qux", "foo": "bar"}, [{"op": "remove", "path": "/baz"}], {"foo": "bar"}, False), ("A.4 remove array element", {"foo": ["bar", "qux", "baz"]}, [{"op": "remove", "path": "/foo/1"}], {"foo": ["bar", "baz"]}, False), ("A.5 replace value", {"baz": "qux", "foo": "bar"}, [{"op": "replace", "path": "/baz", "value": "boo"}], {"baz": "boo", "foo": "bar"}, False), ("A.6 replace missing -> error", {"foo": "bar"}, [{"op": "replace", "path": "/baz", "value": "qux"}], None, True), ("A.7 move value", {"foo": {"bar": "baz", "waldo": "fred"}, "qux": {"corge": "grault"}}, [{"op": "move", "from": "/foo/waldo", "path": "/qux/thud"}], {"foo": {"bar": "baz"}, "qux": {"corge": "grault", "thud": "fred"}}, False), ("A.8 move array element", {"foo": ["all", "grass", "cows", "eat"]}, [{"op": "move", "from": "/foo/1", "path": "/foo/3"}], {"foo": ["all", "cows", "eat", "grass"]}, False), ("A.9 test success", {"baz": "qux", "foo": ["a", 2, "c"]}, [{"op": "test", "path": "/baz", "value": "qux"}], {"baz": "qux", "foo": ["a", 2, "c"]}, False), ("A.10 test failure -> error", {"baz": "qux"}, [{"op": "test", "path": "/baz", "value": "bar"}], None, True), ("A.11 add to root", {"foo": "bar"}, [{"op": "add", "path": "", "value": {"baz": "qux"}}], {"baz": "qux"}, False), ("A.12 escaped ~1", {"": "/", "/": " slash "}, [{"op": "test", "path": "/~1", "value": " slash "}], {"": "/", "/": " slash "}, False), ("A.13 escaped ~0", {"~": "tilde", "/": "slash"}, [{"op": "test", "path": "/~0", "value": "tilde"}], {"~": "tilde", "/": "slash"}, False), ("A.14 add out of range -> error", {"foo": ["bar", "qux", "baz"]}, [{"op": "add", "path": "/foo/5", "value": "qux"}], None, True), ("A.15 append with '-'", {"foo": ["bar", "baz"]}, [{"op": "add", "path": "/foo/-", "value": "qux"}], {"foo": ["bar", "baz", "qux"]}, False), ("A.16 add into existing nested", {"foo": {"bar": "baz"}}, [{"op": "add", "path": "/foo/bar", "value": "qux"}], {"foo": {"bar": "qux"}}, False), ("A.17 test array element", {"foo": ["bar", "baz"]}, [{"op": "test", "path": "/foo/1", "value": "baz"}], {"foo": ["bar", "baz"]}, False), ("A.18 test object", {"foo": ["bar", "baz"]}, [{"op": "test", "path": "/foo", "value": ["bar", "baz"]}], {"foo": ["bar", "baz"]}, False), ("A.19 test nested value", {"foo": {"bar": {"baz": "qux"}}}, [{"op": "test", "path": "/foo/bar/baz", "value": "qux"}], {"foo": {"bar": {"baz": "qux"}}}, False), ("A.20 add into nested array", {"foo": [{"bar": "baz"}]}, [{"op": "add", "path": "/foo/0/bar", "value": "qux"}], {"foo": [{"bar": "qux"}]}, False), ("A.21 test failure -> error", {"foo": ["bar", "baz"]}, [{"op": "test", "path": "/foo/1", "value": "qux"}], None, True), # ---- spec_tests.json extras ---- ("spec 4.1 add with missing object -> error", {"q": {"bar": 2}}, [{"op": "add", "path": "/a/b", "value": 1}], None, True), ("spec A.12 add to non-existent target -> error", {"foo": "bar"}, [{"op": "add", "path": "/baz/bat", "value": "qux"}], None, True), ("spec A.14 ~ escape ordering", {"/": 9, "~1": 10}, [{"op": "test", "path": "/~01", "value": 10}], {"/": 9, "~1": 10}, False), ("spec A.15 number vs string -> error", {"/": 9, "~1": 10}, [{"op": "test", "path": "/~01", "value": "10"}], None, True), ("spec A.16 append array value", {"foo": ["bar"]}, [{"op": "add", "path": "/foo/-", "value": ["abc", "def"]}], {"foo": ["bar", ["abc", "def"]]}, False), # ---- RFC 6901 pointer quirks ---- ("6901 whole document pointer is ''", {"a": 1}, [{"op": "test", "path": "", "value": {"a": 1}}], {"a": 1}, False), ("6901 leading-zero index rejected", {"foo": ["a", "b"]}, [{"op": "remove", "path": "/foo/01"}], None, True), ("6901 '-' is not a read index", {"foo": ["a"]}, [{"op": "remove", "path": "/foo/-"}], None, True), ("6901 '~2' escape rejected", {"foo": 1}, [{"op": "test", "path": "/fo~2o", "value": 1}], None, True), ("6901 dangling '~' rejected", {"foo": 1}, [{"op": "test", "path": "/foo~", "value": 1}], None, True), ("6901 pointer must start with '/'", {"foo": 1}, [{"op": "test", "path": "foo", "value": 1}], None, True), ("6901 path must be a string", {"foo": 1}, [{"op": "test", "path": None, "value": 1}], None, True), # ---- operation semantics ---- ("move /a -> /a is a no-op", {"a": 1, "b": 2}, [{"op": "move", "from": "/a", "path": "/a"}], {"a": 1, "b": 2}, False), ("move into own descendant -> error", {"a": {"b": 1}}, [{"op": "move", "from": "/a", "path": "/a/b"}], None, True), ("copy array element into object", {"foo": ["a", "b"]}, [{"op": "copy", "from": "/foo/0", "path": "/bar"}], {"foo": ["a", "b"], "bar": "a"}, False), ("copy into array middle", {"foo": [1, 2, 3]}, [{"op": "copy", "from": "/foo/0", "path": "/foo/1"}], {"foo": [1, 1, 2, 3]}, False), ("add at index == len is legal", {"foo": [1, 2]}, [{"op": "add", "path": "/foo/2", "value": 3}], {"foo": [1, 2, 3]}, False), ("replace whole document", {"a": 1}, [{"op": "replace", "path": "", "value": [1, 2]}], [1, 2], False), ("remove whole document -> error", {"a": 1}, [{"op": "remove", "path": ""}], None, True), ("empty patch is a no-op", {"a": 1}, [], {"a": 1}, False), ("chained ops on the live doc", {"a": 1}, [{"op": "add", "path": "/b", "value": 2}, {"op": "remove", "path": "/a"}, {"op": "replace", "path": "/b", "value": 3}], {"b": 3}, False), ("unknown op rejected", {"a": 1}, [{"op": "spam", "path": "/a", "value": 1}], None, True), ("missing value rejected", {"a": 1}, [{"op": "add", "path": "/b"}], None, True), ("missing path rejected", {"a": 1}, [{"op": "remove"}], None, True), ("patch must be an array", {"a": 1}, {"op": "add", "path": "/b", "value": 2}, None, True), ("non-object patch entry rejected", {"a": 1}, ["add"], None, True), # ---- test-op equality ---- ("test: key order irrelevant", {"b": 1, "a": 2}, [{"op": "test", "path": "", "value": {"a": 2, "b": 1}}], {"b": 1, "a": 2}, False), ("test: 1 == 1.0", {"a": 1}, [{"op": "test", "path": "/a", "value": 1.0}], {"a": 1}, False), ("test: true != 1", {"a": True}, [{"op": "test", "path": "/a", "value": 1}], None, True), ("test: false != 0", {"a": False}, [{"op": "test", "path": "/a", "value": 0}], None, True), ("test: null == null", {"a": None}, [{"op": "test", "path": "/a", "value": None}], {"a": None}, False), ("test: array order significant", {"a": [1, 2]}, [{"op": "test", "path": "/a", "value": [2, 1]}], None, True), ("test: deep-copied value is not shared", {"a": {"x": 1}}, [{"op": "copy", "from": "/a", "path": "/b"}], {"a": {"x": 1}, "b": {"x": 1}}, False), ] def run_selftest(verbose: bool = False) -> int: """RFC 6902 Appendix A + RFC 6901 + equality semantics. 0 = all pass.""" passed = failed = 0 for name, doc, patch, expected, must_raise in _selftest_cases(): try: got = apply_patch(doc, patch) except JsonPatchError as exc: if must_raise: passed += 1 if verbose: print(f" PASS {name} (raised: {exc})") else: failed += 1 print(f" FAIL {name}: unexpected JsonPatchError: {exc}") continue except Exception as exc: # noqa: BLE001 - surface, never mask failed += 1 print(f" FAIL {name}: unexpected {type(exc).__name__}: {exc}") continue if must_raise: failed += 1 print(f" FAIL {name}: expected JsonPatchError, got {got!r}") elif json_equal(got, expected): passed += 1 if verbose: print(f" PASS {name}") else: failed += 1 print(f" FAIL {name}: expected {_canonical(expected)!r}, " f"got {_canonical(got)!r}") print(f"\n== {passed} passed, {failed} failed ==") return 0 if failed == 0 else 1 def _selftest_json() -> int: doc_s = '{"foo": ["bar", "baz"], "n": 1}' patch_s = ('[{"op":"add","path":"/foo/1","value":"qux"},' '{"op":"test","path":"/n","value":1}]') out_s = apply_patch_text(doc_s, patch_s) want = '{"foo": ["bar", "qux", "baz"], "n": 1}' if out_s == want: print(f"json-selftest PASS -> {out_s}") return 0 print(f"json-selftest FAIL: {out_s!r} != {want!r}") return 1 def main(argv=None) -> int: argv = list(sys.argv[1:] if argv is None else argv) if "--selftest" in argv: return run_selftest(verbose=("-v" in argv or "--verbose" in argv)) if "--selftest-json" in argv: return _selftest_json() doc_s = patch_s = None i = 0 while i < len(argv): if argv[i] == "--doc" and i + 1 < len(argv): doc_s = argv[i + 1] i += 2 elif argv[i] == "--patch" and i + 1 < len(argv): patch_s = argv[i + 1] i += 2 else: print(f"usage: {sys.argv[0]} [--selftest [--verbose]] " "[--selftest-json] | --doc FILE --patch FILE", file=sys.stderr) return USAGE_ERR if doc_s is None or patch_s is None: print("usage: --doc FILE --patch FILE (or --selftest)", file=sys.stderr) return USAGE_ERR def _read(spec: str) -> str: return sys.stdin.read() if spec == "-" else open(spec, encoding="utf-8").read() try: print(apply_patch_text(_read(doc_s), _read(patch_s))) except (JsonPatchError, json.JSONDecodeError, OSError) as exc: print(f"error: {exc}", file=sys.stderr) return PATCH_ERR return 0 if __name__ == "__main__": sys.exit(main())