Atomare JSON-Patch-Anwendung in Python implementieren

import copy

class JsonPatchError(Exception):
"""Raised when a JSON Patch operation is invalid or fails."""
pass

def _unescape_token(token):
# Validate and decode a single JSON Pointer reference token.
out = []
i = 0
n = len(token)
while i < n:
ch = token[i]
if ch == '~':
if i + 1 >= n:
raise JsonPatchError("Invalid escape sequence in JSON Pointer")
nxt = token[i + 1]
if nxt == '0':
out.append('~')
elif nxt == '1':
out.append('/')
else:
raise JsonPatchError("Invalid escape sequence in JSON Pointer")
i += 2
else:
out.append(ch)
i += 1
return ''.join(out)

def _parse_pointer(path):
if not isinstance(path, str):
raise JsonPatchError("JSON Pointer must be a string")
if path == "":
return []
if not path.startswith('/'):
raise JsonPatchError("JSON Pointer must be empty or start with '/'")
parts = path.split('/')[1:]
return [_unescape_token(p) for p in parts]

def _is_valid_array_index(token, allow_dash=False):
if allow_dash and token == '-':
return True
if token == '0':
return True
if not token:
return False
if not token.isdigit():
return False
if token[0] == '0':
return False
return True

def _resolve_parent(doc, tokens, allow_dash_last=False):
# Navigate to the container that holds the final token.
# Returns (parent, final_token).
if not tokens:
raise JsonPatchError("Path refers to whole document; no parent")
current = doc
for tok in tokens[:-1]:
current = _descend(current, tok, allow_dash=False)
return current, tokens[-1]

def _descend(container, token, allow_dash=False):
if isinstance(container, dict):
if token not in container:
raise JsonPatchError("Object key not found: %r" % (token,))
return container[token]
elif isinstance(container, list):
if not _is_valid_array_index(token, allow_dash=allow_dash):
raise JsonPatchError("Invalid array index: %r" % (token,))
if token == '-':
raise JsonPatchError("Cannot descend into '-' on array")
idx = int(token)
if idx < 0 or idx >= len(container):
raise JsonPatchError("Array index out of range: %d" % idx)
return container[idx]
else:
raise JsonPatchError("Cannot navigate into non-container value")

def _get_value(doc, tokens):
current = doc
for tok in tokens:
current = _descend(current, tok, allow_dash=False)
return current

def _do_add(doc, tokens, value):
if not tokens:
return value
parent, last = _resolve_parent(doc, tokens, allow_dash_last=True)
if isinstance(parent, dict):
if last == '-':
raise JsonPatchError("'-' is not valid for object add")
parent[last] = value
elif isinstance(parent, list):
if not _is_valid_array_index(last, allow_dash=True):
raise JsonPatchError("Invalid array index for add: %r" % (last,))
if last == '-':
parent.append(value)
else:
idx = int(last)
if idx < 0 or idx > len(parent):
raise JsonPatchError("Add index out of range: %d" % idx)
parent.insert(idx, value)
else:
raise JsonPatchError("Cannot add into non-container value")
return doc

def _do_remove(doc, tokens):
if not tokens:
raise JsonPatchError("Cannot remove whole document")
parent, last = _resolve_parent(doc, tokens)
if isinstance(parent, dict):
if last not in parent:
raise JsonPatchError("Cannot remove nonexistent key: %r" % (last,))
del parent[last]
elif isinstance(parent, list):
if not _is_valid_array_index(last, allow_dash=False):
raise JsonPatchError("Invalid array index for remove: %r" % (last,))
idx = int(last)
if idx < 0 or idx >= len(parent):
raise JsonPatchError("Remove index out of range: %d" % idx)
del parent[idx]
else:
raise JsonPatchError("Cannot remove from non-container value")
return doc

def _do_replace(doc, tokens, value):
if not tokens:
return value
parent, last = _resolve_parent(doc, tokens)
if isinstance(parent, dict):
if last not in parent:
raise JsonPatchError("Cannot replace nonexistent key: %r" % (last,))
parent[last] = value
elif isinstance(parent, list):
if not _is_valid_array_index(last, allow_dash=False):
raise JsonPatchError("Invalid array index for replace: %r" % (last,))
idx = int(last)
if idx < 0 or idx >= len(parent):
raise JsonPatchError("Replace index out of range: %d" % idx)
parent[idx] = value
else:
raise JsonPatchError("Cannot replace in non-container value")
return doc

def _deep_equal(a, b):
# Exact type-aware equality. Disallow bool/number cross matches.
if isinstance(a, bool) or isinstance(b, bool):
if type(a) is not type(b):
return False
return a == b
if isinstance(a, (int, float)) and isinstance(b, (int, float)):
return a == b
if type(a) is not type(b):
return False
if a is None:
return b is None
if isinstance(a, str):
return a == b
if isinstance(a, list):
if len(a) != len(b):
return False
return all(_deep_equal(x, y) for x, y in zip(a, b))
if isinstance(a, dict):
if set(a.keys()) != set(b.keys()):
return False
return all(_deep_equal(a[k], b[k]) for k in a)
return a == b

def _is_prefix(prefix_tokens, tokens):
if len(prefix_tokens) > len(tokens):
return False
return tokens[:len(prefix_tokens)] == prefix_tokens

_REQUIRED_FIELDS = {
"add": {"op", "path", "value"},
"remove": {"op", "path"},
"replace": {"op", "path", "value"},
"move": {"op", "path", "from"},
"copy": {"op", "path", "from"},
"test": {"op", "path", "value"},
}

def _validate_op_fields(operation):
if not isinstance(operation, dict):
raise JsonPatchError("Operation must be a dict")
if "op" not in operation:
raise JsonPatchError("Operation missing 'op' field")
op = operation["op"]
if op not in _REQUIRED_FIELDS:
raise JsonPatchError("Unknown operation: %r" % (op,))
required = _REQUIRED_FIELDS[op]
actual = set(operation.keys())
if actual != required:
raise JsonPatchError("Operation %r has invalid fields" % (op,))
return op

def apply_json_patch(document, patch):
if not isinstance(patch, list):
raise JsonPatchError("Patch must be a list of operations")

# Work on a deep copy so the original is never mutated.
working = copy.deepcopy(document)

for operation in patch:
    op = _validate_op_fields(operation)

    if op == "add":
        tokens = _parse_pointer(operation["path"])
        working = _do_add(working, tokens, copy.deepcopy(operation["value"]))

    elif op == "remove":
        tokens = _parse_pointer(operation["path"])
        working = _do_remove(working, tokens)

    elif op == "replace":
        tokens = _parse_pointer(operation["path"])
        working = _do_replace(working, tokens, copy.deepcopy(operation["value"]))

    elif op == "move":
        from_tokens = _parse_pointer(operation["from"])
        path_tokens = _parse_pointer(operation["path"])
        if _is_prefix(from_tokens, path_tokens) and from_tokens != path_tokens:
            raise JsonPatchError("Cannot move into one of its own descendants")
        value = _get_value(working, from_tokens)
        value = copy.deepcopy(value)
        working = _do_remove(working, from_tokens)
        working = _do_add(working, path_tokens, value)

    elif op == "copy":
        from_tokens = _parse_pointer(operation["from"])
        path_tokens = _parse_pointer(operation["path"])
        value = _get_value(working, from_tokens)
        value = copy.deepcopy(value)
        working = _do_add(working, path_tokens, value)

    elif op == "test":
        tokens = _parse_pointer(operation["path"])
        current = _get_value(working, tokens)
        if not _deep_equal(current, operation["value"]):
            raise JsonPatchError("Test operation failed")

return working

import copy

class JsonPatchError(Exception):
"""Benutzerdefinierte Ausnahme für JSON-Patch-Fehler."""
pass

def _decode_token(token: str) -> str:
"""Dekodiert ein JSON-Pointer-Token und verarbeitet dabei die Escape-Sequenzen ~0 und ~1."""
if '~' in token and token.replace('~0', '').replace('~1', '').find('~') != -1:
raise JsonPatchError(f"Ungültige Escape-Sequenz im Token: {token}")
return token.replace('~1', '/').replace('~0', '~')

def _parse_pointer(path: str) -> list[str]:
"""Parst einen JSON-Pointer-String in eine Liste dekodierter Tokens."""
if not isinstance(path, str):
raise JsonPatchError("Der JSON-Pointer-Pfad muss ein String sein.")
if path == "":
return []
if not path.startswith('/'):
raise JsonPatchError(f"Ungültiger JSON-Pointer-Pfad: {path}")
return [_decode_token(token) for token in path[1:].split('/')]

def _parse_array_index_token(token: str) -> int:
"""Parst ein String-Token in eine gültige Ganzzahl für einen Array-Index."""
if not token.isdigit():
raise JsonPatchError(f"Ungültiger Array-Index: '{token}'")
if len(token) > 1 and token.startswith('0'):
raise JsonPatchError(f"Ungültiger Array-Index mit führender Null: '{token}'")
return int(token)

def _get_parent_and_key(doc, tokens: list[str]):
"""
Durchläuft das Dokument, um den übergeordneten Container und den endgültigen Schlüssel/Index zu finden.
Gibt (parent_container, final_token) zurück.
Löst JsonPatchError aus, wenn der Pfad ungültig ist.
"""
if not tokens:
raise JsonPatchError("Der übergeordnete Knoten des Wurzeldokuments kann nicht ermittelt werden.")

parent = doc
for token in tokens[:-1]:
    if isinstance(parent, dict):
        if token not in parent:
            raise JsonPatchError(f"Pfadsegment nicht gefunden: '{token}'")
        parent = parent[token]
    elif isinstance(parent, list):
        try:
            index = _parse_array_index_token(token)
            parent = parent[index]
        except (JsonPatchError, IndexError):
            raise JsonPatchError(f"Pfadsegment nicht gefunden: '{token}'")
    else:
        raise JsonPatchError(f"Containerlosen Typ {type(parent).__name__} kann nicht durchlaufen werden.")
        
return parent, tokens[-1]

def _get_value_at_path(doc, path_str: str):
"""
Ruft den Wert an einem gegebenen JSON-Pointer-Pfad ab.
Löst JsonPatchError aus, wenn der Pfad nicht existiert.
"""
tokens = _parse_pointer(path_str)
if not tokens:
return doc

try:
    parent, key = _get_parent_and_key(doc, tokens)
    if isinstance(parent, dict):
        return parent[key]
    elif isinstance(parent, list):
        index = _parse_array_index_token(key)
        return parent[index]
    else:
        raise JsonPatchError("Ungültige Pfaddurchquerung.")
except (KeyError, IndexError, JsonPatchError):
    raise JsonPatchError(f"Pfad existiert nicht: {path_str}")

def _validate_op_keys(op_dict, required_keys, op_index):
"""Prüft, dass das Operations-Wörterbuch genau die erforderlichen Schlüssel enthält."""
op_keys = set(op_dict.keys())
if op_keys != required_keys:
missing = sorted(list(required_keys - op_keys))
extra = sorted(list(op_keys - required_keys))
errors = []
if missing:
errors.append(f"fehlende Schlüssel: {missing}")
if extra:
errors.append(f"unbekannte Schlüssel: {extra}")
raise JsonPatchError(f"Ungültiges Operationsobjekt an Index {op_index}: {', '.join(errors)}")

def apply_json_patch(document, patch):
"""
Wendet eine JSON-Patch-artige Folge von Operationen auf einen JSON-kompatiblen Wert an.

Die Implementierung verändert das ursprüngliche Dokument nicht. Wenn eine Operation
ungültig ist, löst sie eine JsonPatchError aus und lässt das Dokument unverändert.

:param document: Das JSON-kompatible Dokument (dict, list usw.).
:param patch: Eine Liste von Wörterbüchern mit JSON-Patch-Operationen.
:return: Ein neues, gepatchtes Dokument.
:raises JsonPatchError: Wenn eine Operation ungültig ist.
"""
if not isinstance(patch, list):
    raise JsonPatchError("Patch muss eine Liste von Operationen sein.")

new_doc = copy.deepcopy(document)

for i, op_dict in enumerate(patch):
    if not isinstance(op_dict, dict):
        raise JsonPatchError(f"Operation an Index {i} ist kein Wörterbuch.")

    op = op_dict.get('op')
    if op not in ('add', 'remove', 'replace', 'move', 'copy', 'test'):
        raise JsonPatchError(f"Unbekanntes oder fehlendes 'op' an Index {i}.")

    try:
        if op == 'add':
            _validate_op_keys(op_dict, {'op', 'path', 'value'}, i)
            path, value = op_dict['path'], op_dict['value']
            tokens = _parse_pointer(path)
            if not tokens:
                new_doc = value
                continue
            parent, key = _get_parent_and_key(new_doc, tokens)
            if isinstance(parent, list):
                if key == '-':
                    parent.append(value)
                else:
                    index = _parse_array_index_token(key)
                    if index > len(parent):
                        raise JsonPatchError(f"Index {index} außerhalb des gültigen Bereichs für add.")
                    parent.insert(index, value)
            elif isinstance(parent, dict):
                parent[key] = value
            else:
                raise JsonPatchError("Ziel für 'add' ist weder eine Liste noch ein Wörterbuch.")

        elif op == 'remove':
            _validate_op_keys(op_dict, {'op', 'path'}, i)
            path = op_dict['path']
            tokens = _parse_pointer(path)
            if not tokens:
                raise JsonPatchError("Das Wurzeldokument kann nicht entfernt werden.")
            parent, key = _get_parent_and_key(new_doc, tokens)
            if isinstance(parent, list):
                index = _parse_array_index_token(key)
                if index >= len(parent):
                    raise JsonPatchError("Index außerhalb des gültigen Bereichs für remove.")
                del parent[index]
            elif isinstance(parent, dict):
                if key not in parent:
                    raise JsonPatchError("Schlüssel für remove nicht gefunden.")
                del parent[key]
            else:
                raise JsonPatchError("Ziel für 'remove' ist weder eine Liste noch ein Wörterbuch.")

        elif op == 'replace':
            _validate_op_keys(op_dict, {'op', 'path', 'value'}, i)
            path, value = op_dict['path'], op_dict['value']
            tokens = _parse_pointer(path)
            if not tokens:
                new_doc = value
                continue
            parent, key = _get_parent_and_key(new_doc, tokens)
            if isinstance(parent, list):
                index = _parse_array_index_token(key)
                if index >= len(parent):
                    raise JsonPatchError("Index außerhalb des gültigen Bereichs für replace.")
                parent[index] = value
            elif isinstance(parent, dict):
                if key not in parent:
                    raise JsonPatchError("Schlüssel für replace nicht gefunden.")
                parent[key] = value
            else:
                raise JsonPatchError("Ziel für 'replace' ist weder eine Liste noch ein Wörterbuch.")

        elif op == 'test':
            _validate_op_keys(op_dict, {'op', 'path', 'value'}, i)
            path, expected_value = op_dict['path'], op_dict['value']
            actual_value = _get_value_at_path(new_doc, path)
            if actual_value != expected_value:
                raise JsonPatchError("Wert stimmt nicht überein.")

        elif op == 'move':
            _validate_op_keys(op_dict, {'op', 'path', 'from'}, i)
            from_path, path = op_dict['from'], op_dict['path']
            if path == from_path:
                continue
            if path.startswith(from_path + '/') and from_path:
                raise JsonPatchError("Ein Objekt kann nicht in einen seiner eigenen Nachfolger verschoben werden.")
            
            from_tokens = _parse_pointer(from_path)
            if not from_tokens:
                raise JsonPatchError("Das Wurzeldokument kann nicht verschoben werden.")
            from_parent, from_key = _get_parent_and_key(new_doc, from_tokens)
            if isinstance(from_parent, dict):
                if from_key not in from_parent:
                    raise JsonPatchError(f"'from'-Pfad existiert nicht: {from_path}")
                value_to_move = from_parent.pop(from_key)
            elif isinstance(from_parent, list):
                from_index = _parse_array_index_token(from_key)
                if from_index >= len(from_parent):
                    raise JsonPatchError(f"'from'-Pfad existiert nicht: {from_path}")
                value_to_move = from_parent.pop(from_index)
            else:
                raise JsonPatchError("Quelle für 'move' ist weder eine Liste noch ein Wörterbuch.")

            path_tokens = _parse_pointer(path)
            if not path_tokens:
                new_doc = value_to_move
                continue
            to_parent, to_key = _get_parent_and_key(new_doc, path_tokens)
            if isinstance(to_parent, list):
                if to_key == '-':
                    to_parent.append(value_to_move)
                else:
                    to_index = _parse_array_index_token(to_key)
                    if to_index > len(to_parent):
                        raise JsonPatchError(f"Index {to_index} außerhalb des gültigen Bereichs für das move-Ziel.")
                    to_parent.insert(to_index, value_to_move)
            elif isinstance(to_parent, dict):
                to_parent[to_key] = value_to_move
            else:
                raise JsonPatchError("Ziel für 'move' ist weder eine Liste noch ein Wörterbuch.")

        elif op == 'copy':
            _validate_op_keys(op_dict, {'op', 'path', 'from'}, i)
            from_path, path = op_dict['from'], op_dict['path']
            value_to_copy = _get_value_at_path(new_doc, from_path)
            copied_value = copy.deepcopy(value_to_copy)

            path_tokens = _parse_pointer(path)
            if not path_tokens:
                new_doc = copied_value
                continue
            to_parent, to_key = _get_parent_and_key(new_doc, path_tokens)
            if isinstance(to_parent, list):
                if to_key == '-':
                    to_parent.append(copied_value)
                else:
                    to_index = _parse_array_index_token(to_key)
                    if to_index > len(to_parent):
                        raise JsonPatchError(f"Index {to_index} außerhalb des gültigen Bereichs für das copy-Ziel.")
                    to_parent.insert(to_index, copied_value)
            elif isinstance(to_parent, dict):
                to_parent[to_key] = copied_value
            else:
                raise JsonPatchError("Ziel für 'copy' ist weder eine Liste noch ein Wörterbuch.")
    
    except JsonPatchError as e:
        raise JsonPatchError(f"Operation an Index {i} ({op}) fehlgeschlagen: {e}") from e
    except (KeyError, IndexError) as e:
        raise JsonPatchError(f"Operation an Index {i} ({op}) fehlgeschlagen: Pfad existiert nicht.") from e

return new_doc