"""Small, dependency-free RFC 6902 JSON Patch implementation for CDSL revisions.""" from __future__ import annotations from copy import deepcopy from typing import Any class CdslPatchError(ValueError): """A JSON Patch operation cannot be applied to the requested CDSL revision.""" def _tokens(path: Any) -> list[str]: if not isinstance(path, str) or not path.startswith("/"): raise CdslPatchError("JSON Patch paths must be JSON Pointers beginning with '/'") return [token.replace("~1", "/").replace("~0", "~") for token in path[1:].split("/")] def _index(token: str, length: int, *, allow_append: bool = False) -> int: if allow_append and token == "-": return length if not token.isdigit() or (len(token) > 1 and token.startswith("0")): raise CdslPatchError(f"Invalid JSON Patch array index: {token}") value = int(token) if value < 0 or value >= length: raise CdslPatchError(f"JSON Patch array index is out of range: {token}") return value def _parent(document: Any, path: str) -> tuple[Any, str]: if path == "": raise CdslPatchError("Replacing the complete CDSL document is not allowed; call generate_cdsl_model instead") tokens = _tokens(path) current = document for token in tokens[:-1]: if isinstance(current, dict): if token not in current: raise CdslPatchError(f"JSON Patch path does not exist: {path}") current = current[token] elif isinstance(current, list): current = current[_index(token, len(current))] else: raise CdslPatchError(f"JSON Patch path does not resolve to a container: {path}") return current, tokens[-1] def _get(document: Any, path: str) -> Any: current = document for token in _tokens(path): if isinstance(current, dict): if token not in current: raise CdslPatchError(f"JSON Patch path does not exist: {path}") current = current[token] elif isinstance(current, list): current = current[_index(token, len(current))] else: raise CdslPatchError(f"JSON Patch path does not resolve: {path}") return current def _add(document: Any, path: str, value: Any) -> None: parent, token = _parent(document, path) if isinstance(parent, dict): parent[token] = deepcopy(value) elif isinstance(parent, list): index = _index(token, len(parent), allow_append=True) parent.insert(index, deepcopy(value)) else: raise CdslPatchError(f"JSON Patch add target is not a container: {path}") def _remove(document: Any, path: str) -> Any: parent, token = _parent(document, path) if isinstance(parent, dict): if token not in parent: raise CdslPatchError(f"JSON Patch path does not exist: {path}") return parent.pop(token) if isinstance(parent, list): return parent.pop(_index(token, len(parent))) raise CdslPatchError(f"JSON Patch remove target is not a container: {path}") def apply_cdsl_patch(cdsl: dict[str, Any], patches: Any) -> dict[str, Any]: if not isinstance(cdsl, dict): raise CdslPatchError("The base CDSL must be an object") if not isinstance(patches, list) or not patches: raise CdslPatchError("patches must be a non-empty JSON Patch array") if len(patches) > 32: raise CdslPatchError("At most 32 JSON Patch operations are allowed") result: Any = deepcopy(cdsl) for index, operation in enumerate(patches): if not isinstance(operation, dict): raise CdslPatchError(f"Patch operation {index} must be an object") kind = str(operation.get("op") or "") path = operation.get("path") if kind not in {"add", "remove", "replace", "move", "copy", "test"}: raise CdslPatchError(f"Unsupported JSON Patch operation: {kind or ''}") if kind in {"add", "replace", "test"} and "value" not in operation: raise CdslPatchError(f"JSON Patch {kind} requires value") if kind in {"move", "copy"} and "from" not in operation: raise CdslPatchError(f"JSON Patch {kind} requires from") if path == "": raise CdslPatchError("Replacing the complete CDSL document is not allowed; call generate_cdsl_model instead") if kind == "add": _add(result, path, operation["value"]) elif kind == "remove": _remove(result, path) elif kind == "replace": _get(result, path) _remove(result, path) _add(result, path, operation["value"]) elif kind == "move": source = str(operation["from"]) if path == source or str(path).startswith(source.rstrip("/") + "/"): raise CdslPatchError("JSON Patch move cannot move a value into itself") moved = _remove(result, source) _add(result, path, moved) elif kind == "copy": _add(result, path, _get(result, str(operation["from"]))) elif kind == "test" and _get(result, path) != operation["value"]: raise CdslPatchError(f"JSON Patch test failed at {path}") return result