Implementar aplicação atômica de JSON Patch em Python

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):
"""Exceção personalizada para erros de JSON Patch."""
pass

def _decode_token(token: str) -> str:
"""Decodifica um token de JSON Pointer, tratando os escapes ~0 e ~1."""
if '~' in token and token.replace('~0', '').replace('~1', '').find('~') != -1:
raise JsonPatchError(f"Sequência de escape inválida no token: {token}")
return token.replace('~1', '/').replace('~0', '~')

def _parse_pointer(path: str) -> list[str]:
"""Analisa uma string JSON Pointer em uma lista de tokens decodificados."""
if not isinstance(path, str):
raise JsonPatchError("O caminho JSON Pointer deve ser uma string.")
if path == "":
return []
if not path.startswith('/'):
raise JsonPatchError(f"Caminho JSON Pointer inválido: {path}")
return [_decode_token(token) for token in path[1:].split('/')]

def _parse_array_index_token(token: str) -> int:
"""Analisa um token string em um inteiro de índice de array válido."""
if not token.isdigit():
raise JsonPatchError(f"Índice de array inválido: '{token}'")
if len(token) > 1 and token.startswith('0'):
raise JsonPatchError(f"Índice de array inválido com zero à esquerda: '{token}'")
return int(token)

def _get_parent_and_key(doc, tokens: list[str]):
"""
Percorre o documento para encontrar o contêiner pai e a chave/índice final.
Retorna (parent_container, final_token).
Gera JsonPatchError se o caminho for inválido.
"""
if not tokens:
raise JsonPatchError("Não é possível obter o pai do documento raiz.")

parent = doc
for token in tokens[:-1]:
    if isinstance(parent, dict):
        if token not in parent:
            raise JsonPatchError(f"Segmento de caminho não encontrado: '{token}'")
        parent = parent[token]
    elif isinstance(parent, list):
        try:
            index = _parse_array_index_token(token)
            parent = parent[index]
        except (JsonPatchError, IndexError):
            raise JsonPatchError(f"Segmento de caminho não encontrado: '{token}'")
    else:
        raise JsonPatchError(f"Não é possível percorrer um não contêiner do tipo {type(parent).__name__}.")
        
return parent, tokens[-1]

def _get_value_at_path(doc, path_str: str):
"""
Recupera o valor em um determinado caminho JSON Pointer.
Gera JsonPatchError se o caminho não existir.
"""
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("Percurso de caminho inválido.")
except (KeyError, IndexError, JsonPatchError):
    raise JsonPatchError(f"O caminho não existe: {path_str}")

def _validate_op_keys(op_dict, required_keys, op_index):
"""Valida se o dicionário da operação tem exatamente as chaves exigidas."""
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"chaves ausentes: {missing}")
if extra:
errors.append(f"chaves desconhecidas: {extra}")
raise JsonPatchError(f"Objeto de operação inválido no índice {op_index}: {', '.join(errors)}")

def apply_json_patch(document, patch):
"""
Aplica uma sequência de operações no estilo JSON Patch a um valor compatível com JSON.

A implementação não modifica o documento original. Se alguma operação for
inválida, ela gera um JsonPatchError e deixa o documento inalterado.

:param document: O documento compatível com JSON (dict, list, etc.).
:param patch: Uma lista de dicionários de operações JSON Patch.
:return: Um novo documento com o patch aplicado.
:raises JsonPatchError: Se alguma operação for inválida.
"""
if not isinstance(patch, list):
    raise JsonPatchError("Patch deve ser uma lista de operações.")

new_doc = copy.deepcopy(document)

for i, op_dict in enumerate(patch):
    if not isinstance(op_dict, dict):
        raise JsonPatchError(f"A operação no índice {i} não é um dicionário.")

    op = op_dict.get('op')
    if op not in ('add', 'remove', 'replace', 'move', 'copy', 'test'):
        raise JsonPatchError(f"'op' desconhecido ou ausente no índice {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"Índice {index} fora dos limites para add.")
                    parent.insert(index, value)
            elif isinstance(parent, dict):
                parent[key] = value
            else:
                raise JsonPatchError("O destino de 'add' não é uma lista nem um dicionário.")

        elif op == 'remove':
            _validate_op_keys(op_dict, {'op', 'path'}, i)
            path = op_dict['path']
            tokens = _parse_pointer(path)
            if not tokens:
                raise JsonPatchError("Não é possível remover o documento raiz.")
            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("Índice fora dos limites para remove.")
                del parent[index]
            elif isinstance(parent, dict):
                if key not in parent:
                    raise JsonPatchError("Chave não encontrada para remove.")
                del parent[key]
            else:
                raise JsonPatchError("O destino de 'remove' não é uma lista nem um dicionário.")

        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("Índice fora dos limites para replace.")
                parent[index] = value
            elif isinstance(parent, dict):
                if key not in parent:
                    raise JsonPatchError("Chave não encontrada para replace.")
                parent[key] = value
            else:
                raise JsonPatchError("O destino de 'replace' não é uma lista nem um dicionário.")

        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("Incompatibilidade de valor.")

        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("Não é possível mover um objeto para um de seus próprios descendentes.")
            
            from_tokens = _parse_pointer(from_path)
            if not from_tokens:
                raise JsonPatchError("Não é possível mover o documento raiz.")
            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"O caminho 'from' não existe: {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"O caminho 'from' não existe: {from_path}")
                value_to_move = from_parent.pop(from_index)
            else:
                raise JsonPatchError("A origem de 'move' não é uma lista nem um dicionário.")

            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"Índice {to_index} fora dos limites para o destino de move.")
                    to_parent.insert(to_index, value_to_move)
            elif isinstance(to_parent, dict):
                to_parent[to_key] = value_to_move
            else:
                raise JsonPatchError("O destino de 'move' não é uma lista nem um dicionário.")

        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"Índice {to_index} fora dos limites para o destino de copy.")
                    to_parent.insert(to_index, copied_value)
            elif isinstance(to_parent, dict):
                to_parent[to_key] = copied_value
            else:
                raise JsonPatchError("O destino de 'copy' não é uma lista nem um dicionário.")
    
    except JsonPatchError as e:
        raise JsonPatchError(f"A operação no índice {i} ({op}) falhou: {e}") from e
    except (KeyError, IndexError) as e:
        raise JsonPatchError(f"A operação no índice {i} ({op}) falhou: o caminho não existe.") from e

return new_doc