diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/deepdiff/delta.py')
| -rw-r--r-- | .venv/lib/python3.12/site-packages/deepdiff/delta.py | 1217 |
1 files changed, 1217 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/deepdiff/delta.py b/.venv/lib/python3.12/site-packages/deepdiff/delta.py new file mode 100644 index 00000000..a76593cd --- /dev/null +++ b/.venv/lib/python3.12/site-packages/deepdiff/delta.py @@ -0,0 +1,1217 @@ +import copy +import logging +from typing import List, Dict, IO, Callable, Set, Union, Optional +from functools import partial, cmp_to_key +from collections.abc import Mapping +from copy import deepcopy +from deepdiff import DeepDiff +from deepdiff.serialization import pickle_load, pickle_dump +from deepdiff.helper import ( + strings, numbers, + np_ndarray, np_array_factory, numpy_dtypes, get_doc, + not_found, numpy_dtype_string_to_type, dict_, + Opcode, FlatDeltaRow, UnkownValueCode, FlatDataAction, + OPCODE_TAG_TO_FLAT_DATA_ACTION, + FLAT_DATA_ACTION_TO_OPCODE_TAG, + SetOrdered, +) +from deepdiff.path import ( + _path_to_elements, _get_nested_obj, _get_nested_obj_and_force, + GET, GETATTR, parse_path, stringify_path, +) +from deepdiff.anyset import AnySet +from deepdiff.summarize import summarize + +logger = logging.getLogger(__name__) + + +VERIFICATION_MSG = 'Expected the old value for {} to be {} but it is {}. Error found on: {}. You may want to set force=True, especially if this delta is created by passing flat_rows_list or flat_dict_list' +ELEM_NOT_FOUND_TO_ADD_MSG = 'Key or index of {} is not found for {} for setting operation.' +TYPE_CHANGE_FAIL_MSG = 'Unable to do the type change for {} from to type {} due to {}' +VERIFY_BIDIRECTIONAL_MSG = ('You have applied the delta to an object that has ' + 'different values than the original object the delta was made from.') +FAIL_TO_REMOVE_ITEM_IGNORE_ORDER_MSG = 'Failed to remove index[{}] on {}. It was expected to be {} but got {}' +DELTA_NUMPY_OPERATOR_OVERRIDE_MSG = ( + 'A numpy ndarray is most likely being added to a delta. ' + 'Due to Numpy override the + operator, you can only do: delta + ndarray ' + 'and NOT ndarray + delta') +BINIARY_MODE_NEEDED_MSG = "Please open the file in the binary mode and pass to Delta by passing 'b' in open(..., 'b'): {}" +DELTA_AT_LEAST_ONE_ARG_NEEDED = 'At least one of the diff, delta_path or delta_file arguments need to be passed.' +INVALID_ACTION_WHEN_CALLING_GET_ELEM = 'invalid action of {} when calling _get_elem_and_compare_to_old_value' +INVALID_ACTION_WHEN_CALLING_SIMPLE_SET_ELEM = 'invalid action of {} when calling _simple_set_elem_value' +INVALID_ACTION_WHEN_CALLING_SIMPLE_DELETE_ELEM = 'invalid action of {} when calling _simple_set_elem_value' +UNABLE_TO_GET_ITEM_MSG = 'Unable to get the item at {}: {}' +UNABLE_TO_GET_PATH_MSG = 'Unable to get the item at {}' +INDEXES_NOT_FOUND_WHEN_IGNORE_ORDER = 'Delta added to an incompatible object. Unable to add the following items at the specific indexes. {}' +NUMPY_TO_LIST = 'NUMPY_TO_LIST' +NOT_VALID_NUMPY_TYPE = "{} is not a valid numpy type." + +doc = get_doc('delta.rst') + + +class DeltaError(ValueError): + """ + Delta specific errors + """ + pass + + +class DeltaNumpyOperatorOverrideError(ValueError): + """ + Delta Numpy Operator Override Error + """ + pass + + +class Delta: + + __doc__ = doc + + def __init__( + self, + diff: Union[DeepDiff, Mapping, str, bytes, None]=None, + delta_path: Optional[str]=None, + delta_file: Optional[IO]=None, + delta_diff: Optional[dict]=None, + flat_dict_list: Optional[List[Dict]]=None, + flat_rows_list: Optional[List[FlatDeltaRow]]=None, + deserializer: Callable=pickle_load, + log_errors: bool=True, + mutate: bool=False, + raise_errors: bool=False, + safe_to_import: Optional[Set[str]]=None, + serializer: Callable=pickle_dump, + verify_symmetry: Optional[bool]=None, + bidirectional: bool=False, + always_include_values: bool=False, + iterable_compare_func_was_used: Optional[bool]=None, + force: bool=False, + ): + # for pickle deserializer: + if hasattr(deserializer, '__code__') and 'safe_to_import' in set(deserializer.__code__.co_varnames): + _deserializer = deserializer + else: + def _deserializer(obj, safe_to_import=None): + result = deserializer(obj) + if result.get('_iterable_opcodes'): + _iterable_opcodes = {} + for path, op_codes in result['_iterable_opcodes'].items(): + _iterable_opcodes[path] = [] + for op_code in op_codes: + _iterable_opcodes[path].append( + Opcode( + **op_code + ) + ) + result['_iterable_opcodes'] = _iterable_opcodes + return result + + + self._reversed_diff = None + + if verify_symmetry is not None: + logger.warning( + "DeepDiff Deprecation: use bidirectional instead of verify_symmetry parameter." + ) + bidirectional = verify_symmetry + + self.bidirectional = bidirectional + if bidirectional: + self.always_include_values = True # We need to include the values in bidirectional deltas + else: + self.always_include_values = always_include_values + + if diff is not None: + if isinstance(diff, DeepDiff): + self.diff = diff._to_delta_dict(directed=not bidirectional, always_include_values=self.always_include_values) + elif isinstance(diff, Mapping): + self.diff = diff + elif isinstance(diff, strings): + self.diff = _deserializer(diff, safe_to_import=safe_to_import) + elif delta_path: + with open(delta_path, 'rb') as the_file: + content = the_file.read() + self.diff = _deserializer(content, safe_to_import=safe_to_import) + elif delta_diff: + self.diff = delta_diff + elif delta_file: + try: + content = delta_file.read() + except UnicodeDecodeError as e: + raise ValueError(BINIARY_MODE_NEEDED_MSG.format(e)) from None + self.diff = _deserializer(content, safe_to_import=safe_to_import) + elif flat_dict_list: + # Use copy to preserve original value of flat_dict_list in calling module + self.diff = self._from_flat_dicts(copy.deepcopy(flat_dict_list)) + elif flat_rows_list: + self.diff = self._from_flat_rows(copy.deepcopy(flat_rows_list)) + else: + raise ValueError(DELTA_AT_LEAST_ONE_ARG_NEEDED) + + self.mutate = mutate + self.raise_errors = raise_errors + self.log_errors = log_errors + self._numpy_paths = self.diff.get('_numpy_paths', False) + # When we create the delta from a list of flat dictionaries, details such as iterable_compare_func_was_used get lost. + # That's why we allow iterable_compare_func_was_used to be explicitly set. + self._iterable_compare_func_was_used = self.diff.get('_iterable_compare_func_was_used', iterable_compare_func_was_used) + self.serializer = serializer + self.deserializer = deserializer + self.force = force + if force: + self.get_nested_obj = _get_nested_obj_and_force + else: + self.get_nested_obj = _get_nested_obj + self.reset() + + def __repr__(self): + return "<Delta: {}>".format(summarize(self.diff, max_length=100)) + + def reset(self): + self.post_process_paths_to_convert = dict_() + + def __add__(self, other): + if isinstance(other, numbers) and self._numpy_paths: # type: ignore + raise DeltaNumpyOperatorOverrideError(DELTA_NUMPY_OPERATOR_OVERRIDE_MSG) + if self.mutate: + self.root = other + else: + self.root = deepcopy(other) + self._do_pre_process() + self._do_values_changed() + self._do_set_item_added() + self._do_set_item_removed() + self._do_type_changes() + # NOTE: the remove iterable action needs to happen BEFORE + # all the other iterables to match the reverse of order of operations in DeepDiff + self._do_iterable_opcodes() + self._do_iterable_item_removed() + self._do_iterable_item_added() + self._do_ignore_order() + self._do_dictionary_item_added() + self._do_dictionary_item_removed() + self._do_attribute_added() + self._do_attribute_removed() + self._do_post_process() + + other = self.root + # removing the reference to other + del self.root + self.reset() + return other + + __radd__ = __add__ + + def __rsub__(self, other): + if self._reversed_diff is None: + self._reversed_diff = self._get_reverse_diff() + self.diff, self._reversed_diff = self._reversed_diff, self.diff + result = self.__add__(other) + self.diff, self._reversed_diff = self._reversed_diff, self.diff + return result + + def _raise_or_log(self, msg, level='error'): + if self.log_errors: + getattr(logger, level)(msg) + if self.raise_errors: + raise DeltaError(msg) + + def _do_verify_changes(self, path, expected_old_value, current_old_value): + if self.bidirectional and expected_old_value != current_old_value: + if isinstance(path, str): + path_str = path + else: + path_str = stringify_path(path, root_element=('', GETATTR)) + self._raise_or_log(VERIFICATION_MSG.format( + path_str, expected_old_value, current_old_value, VERIFY_BIDIRECTIONAL_MSG)) + + def _get_elem_and_compare_to_old_value( + self, + obj, + path_for_err_reporting, + expected_old_value, + elem=None, + action=None, + forced_old_value=None, + next_element=None, + ): + # if forced_old_value is not None: + try: + if action == GET: + current_old_value = obj[elem] + elif action == GETATTR: + current_old_value = getattr(obj, elem) # type: ignore + else: + raise DeltaError(INVALID_ACTION_WHEN_CALLING_GET_ELEM.format(action)) + except (KeyError, IndexError, AttributeError, TypeError) as e: + if self.force: + if forced_old_value is None: + if next_element is None or isinstance(next_element, str): + _forced_old_value = {} + else: + _forced_old_value = [] + else: + _forced_old_value = forced_old_value + if action == GET: + if isinstance(obj, list): + if isinstance(elem, int) and elem < len(obj): + obj[elem] = _forced_old_value + else: + obj.append(_forced_old_value) + else: + obj[elem] = _forced_old_value + elif action == GETATTR: + setattr(obj, elem, _forced_old_value) # type: ignore + return _forced_old_value + current_old_value = not_found + if isinstance(path_for_err_reporting, (list, tuple)): + path_for_err_reporting = '.'.join([i[0] for i in path_for_err_reporting]) + if self.bidirectional: + self._raise_or_log(VERIFICATION_MSG.format( + path_for_err_reporting, + expected_old_value, current_old_value, e)) + else: + self._raise_or_log(UNABLE_TO_GET_PATH_MSG.format( + path_for_err_reporting)) + return current_old_value + + def _simple_set_elem_value(self, obj, path_for_err_reporting, elem=None, value=None, action=None): + """ + Set the element value directly on an object + """ + try: + if action == GET: + try: + obj[elem] = value + except IndexError: + if elem == len(obj): + obj.append(value) + else: + self._raise_or_log(ELEM_NOT_FOUND_TO_ADD_MSG.format(elem, path_for_err_reporting)) + elif action == GETATTR: + setattr(obj, elem, value) # type: ignore + else: + raise DeltaError(INVALID_ACTION_WHEN_CALLING_SIMPLE_SET_ELEM.format(action)) + except (KeyError, IndexError, AttributeError, TypeError) as e: + self._raise_or_log('Failed to set {} due to {}'.format(path_for_err_reporting, e)) + + def _coerce_obj(self, parent, obj, path, parent_to_obj_elem, + parent_to_obj_action, elements, to_type, from_type): + """ + Coerce obj and mark it in post_process_paths_to_convert for later to be converted back. + Also reassign it to its parent to replace the old object. + """ + self.post_process_paths_to_convert[elements[:-1]] = {'old_type': to_type, 'new_type': from_type} + # If this function is going to ever be used to convert numpy arrays, uncomment these lines: + # if from_type is np_ndarray: + # obj = obj.tolist() + # else: + obj = to_type(obj) + + if parent: + # Making sure that the object is re-instated inside the parent especially if it was immutable + # and we had to turn it into a mutable one. In such cases the object has a new id. + self._simple_set_elem_value(obj=parent, path_for_err_reporting=path, elem=parent_to_obj_elem, + value=obj, action=parent_to_obj_action) + return obj + + def _set_new_value(self, parent, parent_to_obj_elem, parent_to_obj_action, + obj, elements, path, elem, action, new_value): + """ + Set the element value on an object and if necessary convert the object to the proper mutable type + """ + if isinstance(obj, tuple): + # convert this object back to a tuple later + obj = self._coerce_obj( + parent, obj, path, parent_to_obj_elem, + parent_to_obj_action, elements, + to_type=list, from_type=tuple) + if elem != 0 and self.force and isinstance(obj, list) and len(obj) == 0: + # it must have been a dictionary + obj = {} + self._simple_set_elem_value(obj=parent, path_for_err_reporting=path, elem=parent_to_obj_elem, + value=obj, action=parent_to_obj_action) + self._simple_set_elem_value(obj=obj, path_for_err_reporting=path, elem=elem, + value=new_value, action=action) + + def _simple_delete_elem(self, obj, path_for_err_reporting, elem=None, action=None): + """ + Delete the element directly on an object + """ + try: + if action == GET: + del obj[elem] + elif action == GETATTR: + del obj.__dict__[elem] + else: + raise DeltaError(INVALID_ACTION_WHEN_CALLING_SIMPLE_DELETE_ELEM.format(action)) + except (KeyError, IndexError, AttributeError) as e: + self._raise_or_log('Failed to set {} due to {}'.format(path_for_err_reporting, e)) + + def _del_elem(self, parent, parent_to_obj_elem, parent_to_obj_action, + obj, elements, path, elem, action): + """ + Delete the element value on an object and if necessary convert the object to the proper mutable type + """ + obj_is_new = False + if isinstance(obj, tuple): + # convert this object back to a tuple later + self.post_process_paths_to_convert[elements[:-1]] = {'old_type': list, 'new_type': tuple} + obj = list(obj) + obj_is_new = True + self._simple_delete_elem(obj=obj, path_for_err_reporting=path, elem=elem, action=action) + if obj_is_new and parent: + # Making sure that the object is re-instated inside the parent especially if it was immutable + # and we had to turn it into a mutable one. In such cases the object has a new id. + self._simple_set_elem_value(obj=parent, path_for_err_reporting=path, elem=parent_to_obj_elem, + value=obj, action=parent_to_obj_action) + + def _do_iterable_item_added(self): + iterable_item_added = self.diff.get('iterable_item_added', {}) + iterable_item_moved = self.diff.get('iterable_item_moved') + + # First we need to create a placeholder for moved items. + # This will then get replaced below after we go through added items. + # Without this items can get double added because moved store the new_value and does not need item_added replayed + if iterable_item_moved: + added_dict = {v["new_path"]: None for k, v in iterable_item_moved.items()} + iterable_item_added.update(added_dict) + + if iterable_item_added: + self._do_item_added(iterable_item_added, insert=True) + + if iterable_item_moved: + added_dict = {v["new_path"]: v["value"] for k, v in iterable_item_moved.items()} + self._do_item_added(added_dict, insert=False) + + def _do_dictionary_item_added(self): + dictionary_item_added = self.diff.get('dictionary_item_added') + if dictionary_item_added: + self._do_item_added(dictionary_item_added, sort=False) + + def _do_attribute_added(self): + attribute_added = self.diff.get('attribute_added') + if attribute_added: + self._do_item_added(attribute_added) + + @staticmethod + def _sort_key_for_item_added(path_and_value): + elements = _path_to_elements(path_and_value[0]) + # Example elements: [(4.3, 'GET'), ('b', 'GETATTR'), ('a3', 'GET')] + # We only care about the values in the elements not how to get the values. + return [i[0] for i in elements] + + @staticmethod + def _sort_comparison(left, right): + """ + We use sort comparison instead of _sort_key_for_item_added when we run into comparing element types that can not + be compared with each other, such as None to None. Or integer to string. + """ + # Example elements: [(4.3, 'GET'), ('b', 'GETATTR'), ('a3', 'GET')] + # We only care about the values in the elements not how to get the values. + left_path = [i[0] for i in _path_to_elements(left[0], root_element=None)] + right_path = [i[0] for i in _path_to_elements(right[0], root_element=None)] + try: + if left_path < right_path: + return -1 + elif left_path > right_path: + return 1 + else: + return 0 + except TypeError: + if len(left_path) > len(right_path): + left_path = left_path[:len(right_path)] + elif len(right_path) > len(left_path): + right_path = right_path[:len(left_path)] + for l_elem, r_elem in zip(left_path, right_path): + if type(l_elem) != type(r_elem) or type(l_elem) in None: + l_elem = str(l_elem) + r_elem = str(r_elem) + try: + if l_elem < r_elem: + return -1 + elif l_elem > r_elem: + return 1 + except TypeError: + continue + return 0 + + + def _do_item_added(self, items, sort=True, insert=False): + if sort: + # sorting items by their path so that the items with smaller index + # are applied first (unless `sort` is `False` so that order of + # added items is retained, e.g. for dicts). + try: + items = sorted(items.items(), key=self._sort_key_for_item_added) + except TypeError: + items = sorted(items.items(), key=cmp_to_key(self._sort_comparison)) + else: + items = items.items() + + for path, new_value in items: + elem_and_details = self._get_elements_and_details(path) + if elem_and_details: + elements, parent, parent_to_obj_elem, parent_to_obj_action, obj, elem, action = elem_and_details + else: + continue # pragma: no cover. Due to cPython peephole optimizer, this line doesn't get covered. https://github.com/nedbat/coveragepy/issues/198 + + # Insert is only true for iterables, make sure it is a valid index. + if(insert and elem < len(obj)): # type: ignore + obj.insert(elem, None) # type: ignore + + self._set_new_value(parent, parent_to_obj_elem, parent_to_obj_action, + obj, elements, path, elem, action, new_value) + + def _do_values_changed(self): + values_changed = self.diff.get('values_changed') + if values_changed: + self._do_values_or_type_changed(values_changed) + + def _do_type_changes(self): + type_changes = self.diff.get('type_changes') + if type_changes: + self._do_values_or_type_changed(type_changes, is_type_change=True) + + def _do_post_process(self): + if self.post_process_paths_to_convert: + # Example: We had converted some object to be mutable and now we are converting them back to be immutable. + # We don't need to check the change because it is not really a change that was part of the original diff. + self._do_values_or_type_changed(self.post_process_paths_to_convert, is_type_change=True, verify_changes=False) + + def _do_pre_process(self): + if self._numpy_paths and ('iterable_item_added' in self.diff or 'iterable_item_removed' in self.diff): + preprocess_paths = dict_() + for path, type_ in self._numpy_paths.items(): # type: ignore + preprocess_paths[path] = {'old_type': np_ndarray, 'new_type': list} + try: + type_ = numpy_dtype_string_to_type(type_) + except Exception as e: + self._raise_or_log(NOT_VALID_NUMPY_TYPE.format(e)) + continue # pragma: no cover. Due to cPython peephole optimizer, this line doesn't get covered. https://github.com/nedbat/coveragepy/issues/198 + self.post_process_paths_to_convert[path] = {'old_type': list, 'new_type': type_} + if preprocess_paths: + self._do_values_or_type_changed(preprocess_paths, is_type_change=True) + + def _get_elements_and_details(self, path): + try: + elements = _path_to_elements(path) + if len(elements) > 1: + elements_subset = elements[:-2] + if len(elements_subset) != len(elements): + next_element = elements[-2][0] + next2_element = elements[-1][0] + else: + next_element = None + parent = self.get_nested_obj(obj=self, elements=elements_subset, next_element=next_element) + parent_to_obj_elem, parent_to_obj_action = elements[-2] + obj = self._get_elem_and_compare_to_old_value( + obj=parent, path_for_err_reporting=path, expected_old_value=None, + elem=parent_to_obj_elem, action=parent_to_obj_action, next_element=next2_element) # type: ignore + else: + # parent = self + # obj = self.root + # parent_to_obj_elem = 'root' + # parent_to_obj_action = GETATTR + parent = parent_to_obj_elem = parent_to_obj_action = None + obj = self + # obj = self.get_nested_obj(obj=self, elements=elements[:-1]) + elem, action = elements[-1] # type: ignore + except Exception as e: + self._raise_or_log(UNABLE_TO_GET_ITEM_MSG.format(path, e)) + return None + else: + if obj is not_found: + return None + return elements, parent, parent_to_obj_elem, parent_to_obj_action, obj, elem, action + + def _do_values_or_type_changed(self, changes, is_type_change=False, verify_changes=True): + for path, value in changes.items(): + elem_and_details = self._get_elements_and_details(path) + if elem_and_details: + elements, parent, parent_to_obj_elem, parent_to_obj_action, obj, elem, action = elem_and_details + else: + continue # pragma: no cover. Due to cPython peephole optimizer, this line doesn't get covered. https://github.com/nedbat/coveragepy/issues/198 + expected_old_value = value.get('old_value', not_found) + + current_old_value = self._get_elem_and_compare_to_old_value( + obj=obj, path_for_err_reporting=path, expected_old_value=expected_old_value, elem=elem, action=action) + if current_old_value is not_found: + continue # pragma: no cover. I have not been able to write a test for this case. But we should still check for it. + # With type change if we could have originally converted the type from old_value + # to new_value just by applying the class of the new_value, then we might not include the new_value + # in the delta dictionary. That is defined in Model.DeltaResult._from_tree_type_changes + if is_type_change and 'new_value' not in value: + try: + new_type = value['new_type'] + # in case of Numpy we pass the ndarray plus the dtype in a tuple + if new_type in numpy_dtypes: + new_value = np_array_factory(current_old_value, new_type) + else: + new_value = new_type(current_old_value) + except Exception as e: + self._raise_or_log(TYPE_CHANGE_FAIL_MSG.format(obj[elem], value.get('new_type', 'unknown'), e)) # type: ignore + continue + else: + new_value = value['new_value'] + + self._set_new_value(parent, parent_to_obj_elem, parent_to_obj_action, + obj, elements, path, elem, action, new_value) + + if verify_changes: + self._do_verify_changes(path, expected_old_value, current_old_value) + + def _do_item_removed(self, items): + """ + Handle removing items. + """ + # Sorting the iterable_item_removed in reverse order based on the paths. + # So that we delete a bigger index before a smaller index + try: + sorted_item = sorted(items.items(), key=self._sort_key_for_item_added, reverse=True) + except TypeError: + sorted_item = sorted(items.items(), key=cmp_to_key(self._sort_comparison), reverse=True) + for path, expected_old_value in sorted_item: + elem_and_details = self._get_elements_and_details(path) + if elem_and_details: + elements, parent, parent_to_obj_elem, parent_to_obj_action, obj, elem, action = elem_and_details + else: + continue # pragma: no cover. Due to cPython peephole optimizer, this line doesn't get covered. https://github.com/nedbat/coveragepy/issues/198 + + look_for_expected_old_value = False + current_old_value = not_found + try: + if action == GET: + current_old_value = obj[elem] # type: ignore + elif action == GETATTR: + current_old_value = getattr(obj, elem) + look_for_expected_old_value = current_old_value != expected_old_value + except (KeyError, IndexError, AttributeError, TypeError): + look_for_expected_old_value = True + + if look_for_expected_old_value and isinstance(obj, list) and not self._iterable_compare_func_was_used: + # It may return None if it doesn't find it + elem = self._find_closest_iterable_element_for_index(obj, elem, expected_old_value) + if elem is not None: + current_old_value = expected_old_value + if current_old_value is not_found or elem is None: + continue + + self._del_elem(parent, parent_to_obj_elem, parent_to_obj_action, + obj, elements, path, elem, action) + self._do_verify_changes(path, expected_old_value, current_old_value) + + def _find_closest_iterable_element_for_index(self, obj, elem, expected_old_value): + closest_elem = None + closest_distance = float('inf') + for index, value in enumerate(obj): + dist = abs(index - elem) + if dist > closest_distance: + break + if value == expected_old_value and dist < closest_distance: + closest_elem = index + closest_distance = dist + return closest_elem + + def _do_iterable_opcodes(self): + _iterable_opcodes = self.diff.get('_iterable_opcodes', {}) + if _iterable_opcodes: + for path, opcodes in _iterable_opcodes.items(): + transformed = [] + # elements = _path_to_elements(path) + elem_and_details = self._get_elements_and_details(path) + if elem_and_details: + elements, parent, parent_to_obj_elem, parent_to_obj_action, obj, elem, action = elem_and_details + if parent is None: + parent = self + obj = self.root + parent_to_obj_elem = 'root' + parent_to_obj_action = GETATTR + else: + continue # pragma: no cover. Due to cPython peephole optimizer, this line doesn't get covered. https://github.com/nedbat/coveragepy/issues/198 + # import pytest; pytest.set_trace() + obj = self.get_nested_obj(obj=self, elements=elements) + is_obj_tuple = isinstance(obj, tuple) + for opcode in opcodes: + if opcode.tag == 'replace': + # Replace items in list a[i1:i2] with b[j1:j2] + transformed.extend(opcode.new_values) + elif opcode.tag == 'delete': + # Delete items from list a[i1:i2], so we do nothing here + continue + elif opcode.tag == 'insert': + # Insert items from list b[j1:j2] into the new list + transformed.extend(opcode.new_values) + elif opcode.tag == 'equal': + # Items are the same in both lists, so we add them to the result + transformed.extend(obj[opcode.t1_from_index:opcode.t1_to_index]) # type: ignore + if is_obj_tuple: + obj = tuple(obj) # type: ignore + # Making sure that the object is re-instated inside the parent especially if it was immutable + # and we had to turn it into a mutable one. In such cases the object has a new id. + self._simple_set_elem_value(obj=parent, path_for_err_reporting=path, elem=parent_to_obj_elem, + value=obj, action=parent_to_obj_action) + else: + obj[:] = transformed # type: ignore + + + + # obj = self.get_nested_obj(obj=self, elements=elements) + # for + + + def _do_iterable_item_removed(self): + iterable_item_removed = self.diff.get('iterable_item_removed', {}) + + iterable_item_moved = self.diff.get('iterable_item_moved') + if iterable_item_moved: + # These will get added back during items_added + removed_dict = {k: v["value"] for k, v in iterable_item_moved.items()} + iterable_item_removed.update(removed_dict) + + if iterable_item_removed: + self._do_item_removed(iterable_item_removed) + + def _do_dictionary_item_removed(self): + dictionary_item_removed = self.diff.get('dictionary_item_removed') + if dictionary_item_removed: + self._do_item_removed(dictionary_item_removed) + + def _do_attribute_removed(self): + attribute_removed = self.diff.get('attribute_removed') + if attribute_removed: + self._do_item_removed(attribute_removed) + + def _do_set_item_added(self): + items = self.diff.get('set_item_added') + if items: + self._do_set_or_frozenset_item(items, func='union') + + def _do_set_item_removed(self): + items = self.diff.get('set_item_removed') + if items: + self._do_set_or_frozenset_item(items, func='difference') + + def _do_set_or_frozenset_item(self, items, func): + for path, value in items.items(): + elements = _path_to_elements(path) + parent = self.get_nested_obj(obj=self, elements=elements[:-1]) + elem, action = elements[-1] + obj = self._get_elem_and_compare_to_old_value( + parent, path_for_err_reporting=path, expected_old_value=None, elem=elem, action=action, forced_old_value=set()) + new_value = getattr(obj, func)(value) + self._simple_set_elem_value(parent, path_for_err_reporting=path, elem=elem, value=new_value, action=action) + + def _do_ignore_order_get_old(self, obj, remove_indexes_per_path, fixed_indexes_values, path_for_err_reporting): + """ + A generator that gets the old values in an iterable when the order was supposed to be ignored. + """ + old_obj_index = -1 + max_len = len(obj) - 1 + while old_obj_index < max_len: + old_obj_index += 1 + current_old_obj = obj[old_obj_index] + if current_old_obj in fixed_indexes_values: + continue + if old_obj_index in remove_indexes_per_path: + expected_obj_to_delete = remove_indexes_per_path.pop(old_obj_index) + if current_old_obj == expected_obj_to_delete: + continue + else: + self._raise_or_log(FAIL_TO_REMOVE_ITEM_IGNORE_ORDER_MSG.format( + old_obj_index, path_for_err_reporting, expected_obj_to_delete, current_old_obj)) + yield current_old_obj + + def _do_ignore_order(self): + """ + + 't1': [5, 1, 1, 1, 6], + 't2': [7, 1, 1, 1, 8], + + 'iterable_items_added_at_indexes': { + 'root': { + 0: 7, + 4: 8 + } + }, + 'iterable_items_removed_at_indexes': { + 'root': { + 4: 6, + 0: 5 + } + } + + """ + fixed_indexes = self.diff.get('iterable_items_added_at_indexes', dict_()) + remove_indexes = self.diff.get('iterable_items_removed_at_indexes', dict_()) + paths = SetOrdered(fixed_indexes.keys()) | SetOrdered(remove_indexes.keys()) + for path in paths: # type: ignore + # In the case of ignore_order reports, we are pointing to the container object. + # Thus we add a [0] to the elements so we can get the required objects and discard what we don't need. + elem_and_details = self._get_elements_and_details("{}[0]".format(path)) + if elem_and_details: + _, parent, parent_to_obj_elem, parent_to_obj_action, obj, _, _ = elem_and_details + else: + continue # pragma: no cover. Due to cPython peephole optimizer, this line doesn't get covered. https://github.com/nedbat/coveragepy/issues/198 + # copying both these dictionaries since we don't want to mutate them. + fixed_indexes_per_path = fixed_indexes.get(path, dict_()).copy() + remove_indexes_per_path = remove_indexes.get(path, dict_()).copy() + fixed_indexes_values = AnySet(fixed_indexes_per_path.values()) + + new_obj = [] + # Numpy's NdArray does not like the bool function. + if isinstance(obj, np_ndarray): + there_are_old_items = obj.size > 0 + else: + there_are_old_items = bool(obj) + old_item_gen = self._do_ignore_order_get_old( + obj, remove_indexes_per_path, fixed_indexes_values, path_for_err_reporting=path) + while there_are_old_items or fixed_indexes_per_path: + new_obj_index = len(new_obj) + if new_obj_index in fixed_indexes_per_path: + new_item = fixed_indexes_per_path.pop(new_obj_index) + new_obj.append(new_item) + elif there_are_old_items: + try: + new_item = next(old_item_gen) + except StopIteration: + there_are_old_items = False + else: + new_obj.append(new_item) + else: + # pop a random item from the fixed_indexes_per_path dictionary + self._raise_or_log(INDEXES_NOT_FOUND_WHEN_IGNORE_ORDER.format(fixed_indexes_per_path)) + new_item = fixed_indexes_per_path.pop(next(iter(fixed_indexes_per_path))) + new_obj.append(new_item) + + if isinstance(obj, tuple): + new_obj = tuple(new_obj) + # Making sure that the object is re-instated inside the parent especially if it was immutable + # and we had to turn it into a mutable one. In such cases the object has a new id. + self._simple_set_elem_value(obj=parent, path_for_err_reporting=path, elem=parent_to_obj_elem, + value=new_obj, action=parent_to_obj_action) + + def _get_reverse_diff(self): + if not self.bidirectional: + raise ValueError('Please recreate the delta with bidirectional=True') + + SIMPLE_ACTION_TO_REVERSE = { + 'iterable_item_added': 'iterable_item_removed', + 'iterable_items_added_at_indexes': 'iterable_items_removed_at_indexes', + 'attribute_added': 'attribute_removed', + 'set_item_added': 'set_item_removed', + 'dictionary_item_added': 'dictionary_item_removed', + } + # Adding the reverse of the dictionary + for key in list(SIMPLE_ACTION_TO_REVERSE.keys()): + SIMPLE_ACTION_TO_REVERSE[SIMPLE_ACTION_TO_REVERSE[key]] = key + + r_diff = {} + for action, info in self.diff.items(): + reverse_action = SIMPLE_ACTION_TO_REVERSE.get(action) + if reverse_action: + r_diff[reverse_action] = info + elif action == 'values_changed': + r_diff[action] = {} + for path, path_info in info.items(): + reverse_path = path_info['new_path'] if path_info.get('new_path') else path + r_diff[action][reverse_path] = { + 'new_value': path_info['old_value'], 'old_value': path_info['new_value'] + } + elif action == 'type_changes': + r_diff[action] = {} + for path, path_info in info.items(): + reverse_path = path_info['new_path'] if path_info.get('new_path') else path + r_diff[action][reverse_path] = { + 'old_type': path_info['new_type'], 'new_type': path_info['old_type'], + } + if 'new_value' in path_info: + r_diff[action][reverse_path]['old_value'] = path_info['new_value'] + if 'old_value' in path_info: + r_diff[action][reverse_path]['new_value'] = path_info['old_value'] + elif action == 'iterable_item_moved': + r_diff[action] = {} + for path, path_info in info.items(): + old_path = path_info['new_path'] + r_diff[action][old_path] = { + 'new_path': path, 'value': path_info['value'], + } + elif action == '_iterable_opcodes': + r_diff[action] = {} + for path, op_codes in info.items(): + r_diff[action][path] = [] + for op_code in op_codes: + tag = op_code.tag + tag = {'delete': 'insert', 'insert': 'delete'}.get(tag, tag) + new_op_code = Opcode( + tag=tag, + t1_from_index=op_code.t2_from_index, + t1_to_index=op_code.t2_to_index, + t2_from_index=op_code.t1_from_index, + t2_to_index=op_code.t1_to_index, + new_values=op_code.old_values, + old_values=op_code.new_values, + ) + r_diff[action][path].append(new_op_code) + return r_diff + + def dump(self, file): + """ + Dump into file object + """ + # Small optimization: Our internal pickle serializer can just take a file object + # and directly write to it. However if a user defined serializer is passed + # we want to make it compatible with the expectation that self.serializer(self.diff) + # will give the user the serialization and then it can be written to + # a file object when using the dump(file) function. + param_names_of_serializer = set(self.serializer.__code__.co_varnames) + if 'file_obj' in param_names_of_serializer: + self.serializer(self.diff, file_obj=file) + else: + file.write(self.dumps()) + + def dumps(self): + """ + Return the serialized representation of the object as a bytes object, instead of writing it to a file. + """ + return self.serializer(self.diff) + + def to_dict(self): + return dict(self.diff) + + def _flatten_iterable_opcodes(self, _parse_path): + """ + Converts op_codes to FlatDeltaRows + """ + result = [] + for path, op_codes in self.diff['_iterable_opcodes'].items(): + for op_code in op_codes: + result.append( + FlatDeltaRow( + path=_parse_path(path), + action=OPCODE_TAG_TO_FLAT_DATA_ACTION[op_code.tag], + value=op_code.new_values, + old_value=op_code.old_values, + type=type(op_code.new_values), + old_type=type(op_code.old_values), + new_path=None, + t1_from_index=op_code.t1_from_index, + t1_to_index=op_code.t1_to_index, + t2_from_index=op_code.t2_from_index, + t2_to_index=op_code.t2_to_index, + + ) + ) + return result + + @staticmethod + def _get_flat_row(action, info, _parse_path, keys_and_funcs, report_type_changes=True): + for path, details in info.items(): + row = {'path': _parse_path(path), 'action': action} + for key, new_key, func in keys_and_funcs: + if key in details: + if func: + row[new_key] = func(details[key]) + else: + row[new_key] = details[key] + if report_type_changes: + if 'value' in row and 'type' not in row: + row['type'] = type(row['value']) + if 'old_value' in row and 'old_type' not in row: + row['old_type'] = type(row['old_value']) + yield FlatDeltaRow(**row) + + @staticmethod + def _from_flat_rows(flat_rows_list: List[FlatDeltaRow]): + flat_dict_list = (i._asdict() for i in flat_rows_list) + return Delta._from_flat_dicts(flat_dict_list) + + @staticmethod + def _from_flat_dicts(flat_dict_list): + """ + Create the delta's diff object from the flat_dict_list + """ + result = {} + FLATTENING_NEW_ACTION_MAP = { + 'unordered_iterable_item_added': 'iterable_items_added_at_indexes', + 'unordered_iterable_item_removed': 'iterable_items_removed_at_indexes', + } + for flat_dict in flat_dict_list: + index = None + action = flat_dict.get("action") + path = flat_dict.get("path") + value = flat_dict.get('value') + new_path = flat_dict.get('new_path') + old_value = flat_dict.get('old_value', UnkownValueCode) + if not action: + raise ValueError("Flat dict need to include the 'action'.") + if path is None: + raise ValueError("Flat dict need to include the 'path'.") + if action in FLATTENING_NEW_ACTION_MAP: + action = FLATTENING_NEW_ACTION_MAP[action] + index = path.pop() + if action in { + FlatDataAction.attribute_added, + FlatDataAction.attribute_removed, + }: + root_element = ('root', GETATTR) + else: + root_element = ('root', GET) + if isinstance(path, str): + path_str = path + else: + path_str = stringify_path(path, root_element=root_element) # We need the string path + if new_path and new_path != path: + new_path = stringify_path(new_path, root_element=root_element) + else: + new_path = None + if action not in result: + result[action] = {} + if action in { + 'iterable_items_added_at_indexes', + 'iterable_items_removed_at_indexes', + }: + if path_str not in result[action]: + result[action][path_str] = {} + result[action][path_str][index] = value + elif action in { + FlatDataAction.set_item_added, + FlatDataAction.set_item_removed + }: + if path_str not in result[action]: + result[action][path_str] = set() + result[action][path_str].add(value) + elif action in { + FlatDataAction.dictionary_item_added, + FlatDataAction.dictionary_item_removed, + FlatDataAction.attribute_removed, + FlatDataAction.attribute_added, + FlatDataAction.iterable_item_added, + FlatDataAction.iterable_item_removed, + }: + result[action][path_str] = value + elif action == 'values_changed': + if old_value == UnkownValueCode: + result[action][path_str] = {'new_value': value} + else: + result[action][path_str] = {'new_value': value, 'old_value': old_value} + elif action == 'type_changes': + type_ = flat_dict.get('type', UnkownValueCode) + old_type = flat_dict.get('old_type', UnkownValueCode) + + result[action][path_str] = {'new_value': value} + for elem, elem_value in [ + ('new_type', type_), + ('old_type', old_type), + ('old_value', old_value), + ]: + if elem_value != UnkownValueCode: + result[action][path_str][elem] = elem_value + elif action == FlatDataAction.iterable_item_moved: + result[action][path_str] = {'value': value} + elif action in { + FlatDataAction.iterable_items_inserted, + FlatDataAction.iterable_items_deleted, + FlatDataAction.iterable_items_replaced, + FlatDataAction.iterable_items_equal, + }: + if '_iterable_opcodes' not in result: + result['_iterable_opcodes'] = {} + if path_str not in result['_iterable_opcodes']: + result['_iterable_opcodes'][path_str] = [] + result['_iterable_opcodes'][path_str].append( + Opcode( + tag=FLAT_DATA_ACTION_TO_OPCODE_TAG[action], # type: ignore + t1_from_index=flat_dict.get('t1_from_index'), + t1_to_index=flat_dict.get('t1_to_index'), + t2_from_index=flat_dict.get('t2_from_index'), + t2_to_index=flat_dict.get('t2_to_index'), + new_values=flat_dict.get('value'), + old_values=flat_dict.get('old_value'), + ) + ) + if new_path: + result[action][path_str]['new_path'] = new_path + + return result + + def to_flat_dicts(self, include_action_in_path=False, report_type_changes=True) -> List[FlatDeltaRow]: + """ + Returns a flat list of actions that is easily machine readable. + + For example: + {'iterable_item_added': {'root[3]': 5, 'root[2]': 3}} + + Becomes: + [ + {'path': [3], 'value': 5, 'action': 'iterable_item_added'}, + {'path': [2], 'value': 3, 'action': 'iterable_item_added'}, + ] + + + **Parameters** + + include_action_in_path : Boolean, default=False + When False, we translate DeepDiff's paths like root[3].attribute1 into a [3, 'attribute1']. + When True, we include the action to retrieve the item in the path: [(3, 'GET'), ('attribute1', 'GETATTR')] + Note that the "action" here is the different than the action reported by to_flat_dicts. The action here is just about the "path" output. + + report_type_changes : Boolean, default=True + If False, we don't report the type change. Instead we report the value change. + + Example: + t1 = {"a": None} + t2 = {"a": 1} + + dump = Delta(DeepDiff(t1, t2)).dumps() + delta = Delta(dump) + assert t2 == delta + t1 + + flat_result = delta.to_flat_dicts() + flat_expected = [{'path': ['a'], 'action': 'type_changes', 'value': 1, 'new_type': int, 'old_type': type(None)}] + assert flat_expected == flat_result + + flat_result2 = delta.to_flat_dicts(report_type_changes=False) + flat_expected2 = [{'path': ['a'], 'action': 'values_changed', 'value': 1}] + + **List of actions** + + Here are the list of actions that the flat dictionary can return. + iterable_item_added + iterable_item_removed + iterable_item_moved + values_changed + type_changes + set_item_added + set_item_removed + dictionary_item_added + dictionary_item_removed + attribute_added + attribute_removed + """ + return [ + i._asdict() for i in self.to_flat_rows(include_action_in_path=False, report_type_changes=True) + ] # type: ignore + + def to_flat_rows(self, include_action_in_path=False, report_type_changes=True) -> List[FlatDeltaRow]: + """ + Just like to_flat_dicts but returns FlatDeltaRow Named Tuples + """ + result = [] + if include_action_in_path: + _parse_path = partial(parse_path, include_actions=True) + else: + _parse_path = parse_path + if report_type_changes: + keys_and_funcs = [ + ('value', 'value', None), + ('new_value', 'value', None), + ('old_value', 'old_value', None), + ('new_type', 'type', None), + ('old_type', 'old_type', None), + ('new_path', 'new_path', _parse_path), + ] + else: + if not self.always_include_values: + raise ValueError( + "When converting to flat dictionaries, if report_type_changes=False and there are type changes, " + "you must set the always_include_values=True at the delta object creation. Otherwise there is nothing to include." + ) + keys_and_funcs = [ + ('value', 'value', None), + ('new_value', 'value', None), + ('old_value', 'old_value', None), + ('new_path', 'new_path', _parse_path), + ] + + FLATTENING_NEW_ACTION_MAP = { + 'iterable_items_added_at_indexes': 'unordered_iterable_item_added', + 'iterable_items_removed_at_indexes': 'unordered_iterable_item_removed', + } + for action, info in self.diff.items(): + if action == '_iterable_opcodes': + result.extend(self._flatten_iterable_opcodes(_parse_path=_parse_path)) + continue + if action.startswith('_'): + continue + if action in FLATTENING_NEW_ACTION_MAP: + new_action = FLATTENING_NEW_ACTION_MAP[action] + for path, index_to_value in info.items(): + path = _parse_path(path) + for index, value in index_to_value.items(): + path2 = path.copy() + if include_action_in_path: + path2.append((index, 'GET')) # type: ignore + else: + path2.append(index) + if report_type_changes: + row = FlatDeltaRow(path=path2, value=value, action=new_action, type=type(value)) # type: ignore + else: + row = FlatDeltaRow(path=path2, value=value, action=new_action) # type: ignore + result.append(row) + elif action in {'set_item_added', 'set_item_removed'}: + for path, values in info.items(): + path = _parse_path(path) + for value in values: + if report_type_changes: + row = FlatDeltaRow(path=path, value=value, action=action, type=type(value)) + else: + row = FlatDeltaRow(path=path, value=value, action=action) + result.append(row) + elif action == 'dictionary_item_added': + for path, value in info.items(): + path = _parse_path(path) + if isinstance(value, dict) and len(value) == 1: + new_key = next(iter(value)) + path.append(new_key) + value = value[new_key] + elif isinstance(value, (list, tuple)) and len(value) == 1: + value = value[0] + path.append(0) # type: ignore + action = 'iterable_item_added' + elif isinstance(value, set) and len(value) == 1: + value = value.pop() + action = 'set_item_added' + if report_type_changes: + row = FlatDeltaRow(path=path, value=value, action=action, type=type(value)) # type: ignore + else: + row = FlatDeltaRow(path=path, value=value, action=action) # type: ignore + result.append(row) + elif action in { + 'dictionary_item_removed', 'iterable_item_added', + 'iterable_item_removed', 'attribute_removed', 'attribute_added' + }: + for path, value in info.items(): + path = _parse_path(path) + if report_type_changes: + row = FlatDeltaRow(path=path, value=value, action=action, type=type(value)) + else: + row = FlatDeltaRow(path=path, value=value, action=action) + result.append(row) + elif action == 'type_changes': + if not report_type_changes: + action = 'values_changed' + + for row in self._get_flat_row( + action=action, + info=info, + _parse_path=_parse_path, + keys_and_funcs=keys_and_funcs, + report_type_changes=report_type_changes, + ): + result.append(row) + else: + for row in self._get_flat_row( + action=action, + info=info, + _parse_path=_parse_path, + keys_and_funcs=keys_and_funcs, + report_type_changes=report_type_changes, + ): + result.append(row) + return result + + +if __name__ == "__main__": # pragma: no cover + import doctest + doctest.testmod() |