123 lines
5.1 KiB
Python
123 lines
5.1 KiB
Python
"""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 '<missing>'}")
|
|
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
|