op_codes conversion to flat dicts

Author: seperman

deepdiff/delta.py

@@ -11,7 +11,9 @@
     strings, short_repr, numbers,
     np_ndarray, np_array_factory, numpy_dtypes, get_doc,
     not_found, numpy_dtype_string_to_type, dict_,
-    Opcode, FlatDeltaRow, UnkownValueCode,
+    Opcode, FlatDeltaRow, UnkownValueCode, FlatDataAction,
+    OPCODE_TAG_TO_FLAT_DATA_ACTION,
+    FLAT_DATA_ACTION_TO_OPCODE_TAG,
 )
 from deepdiff.path import (
     _path_to_elements, _get_nested_obj, _get_nested_obj_and_force,
@@ -877,6 +879,31 @@ def dumps(self):
     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():
@@ -923,28 +950,44 @@ def _from_flat_dicts(flat_dict_list):
             if action in FLATTENING_NEW_ACTION_MAP:
                 action = FLATTENING_NEW_ACTION_MAP[action]
                 index = path.pop()
-            if action in {'attribute_added', 'attribute_removed'}:
+            if action in {
+                FlatDataAction.attribute_added,
+                FlatDataAction.attribute_removed,
+            }:
                 root_element = ('root', GETATTR)
             else:
                 root_element = ('root', GET)
-            path_str = stringify_path(path, root_element=root_element)  # We need the string path
+            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 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 {'set_item_added', 'set_item_removed'}:
+            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 {
-                'dictionary_item_added', 'dictionary_item_removed',
-                'attribute_removed', 'attribute_added', 'iterable_item_added', 'iterable_item_removed',
+                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':
@@ -964,8 +1007,29 @@ def _from_flat_dicts(flat_dict_list):
                 ]:
                     if elem_value != UnkownValueCode:
                         result[action][path_str][elem] = elem_value
-            elif action == 'iterable_item_moved':
+            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],
+                        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
 
@@ -1066,7 +1130,7 @@ def to_flat_rows(self, include_action_in_path=False, report_type_changes=True) -
         }
         for action, info in self.diff.items():
             if action == '_iterable_opcodes':
-                result.extend(self._flatten_iterable_opcodes())
+                result.extend(self._flatten_iterable_opcodes(_parse_path=_parse_path))
                 continue
             if action.startswith('_'):
                 continue

deepdiff/helper.py

@@ -110,6 +110,17 @@ class pydantic_base_model_type:
 NUMERICS = frozenset(string.digits)
 
 
+class EnumBase(str, enum.Enum):
+    def __repr__(self):
+        """
+        We need to add a single quotes so we can easily copy the value when we do ipdb.
+        """
+        return f"'{self.name}'"
+
+    def __str__(self):
+        return self.name
+
+
 def _int_or_zero(value):
     """
     Tries to extract some number from a string.
@@ -739,6 +750,13 @@ def named_tuple_repr(self):
     return f"{self.__class__.__name__}({', '.join(fields)})"
 
 
+class OpcodeTag(EnumBase):
+    insert = 'insert'
+    delete = 'delete'
+    equal = 'equal'
+    replace = 'replace'
+
+
 class Opcode(NamedTuple):
     tag: str
     t1_from_index: int
@@ -751,7 +769,7 @@ class Opcode(NamedTuple):
     __repr__ = __str__ = named_tuple_repr
 
 
-class FlatDataAction(str, enum.Enum):
+class FlatDataAction(EnumBase):
     values_changed = 'values_changed'
     type_changes = 'type_changes'
     set_item_added = 'set_item_added'
@@ -771,6 +789,16 @@ class FlatDataAction(str, enum.Enum):
     unordered_iterable_item_removed = 'unordered_iterable_item_removed'
 
 
+OPCODE_TAG_TO_FLAT_DATA_ACTION = {
+    OpcodeTag.insert: FlatDataAction.iterable_items_inserted,
+    OpcodeTag.delete: FlatDataAction.iterable_items_deleted,
+    OpcodeTag.replace: FlatDataAction.iterable_items_replaced,
+    OpcodeTag.equal: FlatDataAction.iterable_items_equal,
+}
+
+FLAT_DATA_ACTION_TO_OPCODE_TAG = {v: i for i, v in OPCODE_TAG_TO_FLAT_DATA_ACTION.items()}
+
+
 UnkownValueCode = 'unknown___'
 
 

deepdiff/path.py

@@ -261,7 +261,8 @@ def parse_path(path, root_element=DEFAULT_FIRST_ELEMENT, include_actions=False):
 
     result = _path_to_elements(path, root_element=root_element)
     result = iter(result)
-    next(result)  # We don't want the root item
+    if root_element:
+        next(result)  # We don't want the root item
     if include_actions is False:
         return [i[0] for i in result]
     return [{'element': i[0], 'action': i[1]} for i in result]

tests/test_delta.py

@@ -9,7 +9,7 @@
 from unittest import mock
 from ordered_set import OrderedSet
 from deepdiff import Delta, DeepDiff
-from deepdiff.helper import np, number_to_string, TEXT_VIEW, DELTA_VIEW, CannotCompare, FlatDeltaRow
+from deepdiff.helper import np, number_to_string, TEXT_VIEW, DELTA_VIEW, CannotCompare, FlatDeltaRow, FlatDataAction
 from deepdiff.path import GETATTR, GET
 from deepdiff.delta import (
     ELEM_NOT_FOUND_TO_ADD_MSG,
@@ -2397,6 +2397,29 @@ def test_list_of_alphabet_and_its_delta(self):
         assert l2 == l1 + delta4
         assert l1 == l2 - delta4
 
+        flat_rows = delta2.to_flat_rows()
+
+        expected_flat_rows = [
+            FlatDeltaRow(path=[3], action='values_changed', value='X', old_value='D', type=str, old_type=str, new_path=[2]),
+            FlatDeltaRow(path=[6], action='values_changed', value='Z', old_value='G', type=str, old_type=str),
+            FlatDeltaRow(path=[5], action='values_changed', value='Y', old_value='F', type=str, old_type=str),
+            FlatDeltaRow(path=[], action=FlatDataAction.iterable_items_deleted, value=[], old_value=['A'], type=list, old_type=list, t1_from_index=0, t1_to_index=1, t2_from_index=0, t2_to_index=0),
+            FlatDeltaRow(path=[], action=FlatDataAction.iterable_items_equal, value=None, old_value=None, type=type(None), old_type=type(None), t1_from_index=1, t1_to_index=3, t2_from_index=0, t2_to_index=2),
+            FlatDeltaRow(path=[], action=FlatDataAction.iterable_items_replaced, value=['X'], old_value=['D', 'E', 'F', 'G'], type=list, old_type=list, t1_from_index=3, t1_to_index=7, t2_from_index=2, t2_to_index=3),
+            FlatDeltaRow(path=[], action=FlatDataAction.iterable_items_equal, value=None, old_value=None, type=type(None), old_type=type(None), t1_from_index=7, t1_to_index=9, t2_from_index=3, t2_to_index=5),
+            FlatDeltaRow(path=[], action=FlatDataAction.iterable_items_inserted, value=['Y', 'Z'], old_value=[], type=list, old_type=list, t1_from_index=9, t1_to_index=9, t2_from_index=5, t2_to_index=7)
+        ]
+
+        # The order of the first 3 items is not deterministic
+        assert not DeepDiff(expected_flat_rows[:3], flat_rows[:3], ignore_order=True)
+        assert expected_flat_rows[3:] == flat_rows[3:]
+
+        delta5 = Delta(flat_rows_list=flat_rows, bidirectional=True, force=True)
+
+
+        assert l2 == l1 + delta5
+        assert l1 == l2 - delta5
+
     def test_delta_flat_rows(self):
         t1 = {"key1": "value1"}
         t2 = {"field2": {"key2": "value2"}}