Delta can now read from flat dicts dump

Author: seperman

deepdiff/delta.py

@@ -10,7 +10,10 @@
     np_ndarray, np_array_factory, numpy_dtypes, get_doc,
     not_found, numpy_dtype_string_to_type, dict_,
 )
-from deepdiff.path import _path_to_elements, _get_nested_obj, _get_nested_obj_and_force, GET, GETATTR, parse_path
+from deepdiff.path import (
+    _path_to_elements, _get_nested_obj, _get_nested_obj_and_force,
+    GET, GETATTR, parse_path, stringify_path, DEFAULT_FIRST_ELEMENT
+)
 from deepdiff.anyset import AnySet
 
 
@@ -55,6 +58,10 @@ class DeltaNumpyOperatorOverrideError(ValueError):
     pass
 
 
+class _ObjDoesNotExist:
+    pass
+
+
 class Delta:
 
     __doc__ = doc
@@ -64,6 +71,7 @@ def __init__(
         diff=None,
         delta_path=None,
         delta_file=None,
+        flat_dict_list=None,
         deserializer=pickle_load,
         log_errors=True,
         mutate=False,
@@ -79,6 +87,8 @@ def __init__(
             def _deserializer(obj, safe_to_import=None):
                 return deserializer(obj)
 
+        self._reversed_diff = None
+
         if diff is not None:
             if isinstance(diff, DeepDiff):
                 self.diff = diff._to_delta_dict(directed=not verify_symmetry)
@@ -96,6 +106,8 @@ def _deserializer(obj, safe_to_import=None):
             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:
+            self.diff = self._from_flat_dicts(flat_dict_list)
         else:
             raise ValueError(DELTA_AT_LEAST_ONE_ARG_NEEDED)
 
@@ -161,7 +173,7 @@ def _do_verify_changes(self, path, expected_old_value, current_old_value):
             self._raise_or_log(VERIFICATION_MSG.format(
                 path, expected_old_value, current_old_value, VERIFY_SYMMETRY_MSG))
 
-    def _get_elem_and_compare_to_old_value(self, obj, path_for_err_reporting, expected_old_value, elem=None, action=None):
+    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):
         try:
             if action == GET:
                 current_old_value = obj[elem]
@@ -171,12 +183,12 @@ def _get_elem_and_compare_to_old_value(self, obj, path_for_err_reporting, expect
                 raise DeltaError(INVALID_ACTION_WHEN_CALLING_GET_ELEM.format(action))
         except (KeyError, IndexError, AttributeError, TypeError) as e:
             if self.force:
-                forced_old_value = {}
+                _forced_old_value = {} if forced_old_value is None else forced_old_value
                 if action == GET:
-                    obj[elem] = forced_old_value
+                    obj[elem] = _forced_old_value
                 elif action == GETATTR:
-                    setattr(obj, elem, forced_old_value)
-                return forced_old_value
+                    setattr(obj, elem, _forced_old_value)
+                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])
@@ -475,7 +487,7 @@ def _do_set_or_frozenset_item(self, items, func):
             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)
+                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)
 
@@ -568,6 +580,9 @@ def _do_ignore_order(self):
             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 _reverse_diff(self):
+        pass
+
     def dump(self, file):
         """
         Dump into file object
@@ -604,6 +619,78 @@ def _get_flat_row(action, info, _parse_path, keys_and_funcs):
                         row[new_key] = details[key]
             yield row
 
+    @staticmethod
+    def _from_flat_dicts(flat_dict_list):
+        """
+        Create the delta's diff object from the flat_dict_list
+        """
+        result = {}
+
+        DEFLATTENING_NEW_ACTION_MAP = {
+            'iterable_item_added': 'iterable_items_added_at_indexes',
+            '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')
+            old_value = flat_dict.get('old_value', _ObjDoesNotExist)
+            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 DEFLATTENING_NEW_ACTION_MAP:
+                action = DEFLATTENING_NEW_ACTION_MAP[action]
+                index = path.pop()
+            if action in {'attribute_added', '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 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 {'set_item_added', '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', 'iterable_item_added',
+                'iterable_item_removed', 'attribute_removed', 'attribute_added'
+            }:
+                result[action][path_str] = value
+            elif action == 'values_changed':
+                if old_value is _ObjDoesNotExist:
+                    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', _ObjDoesNotExist)
+                old_type = flat_dict.get('old_type', _ObjDoesNotExist)
+
+                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 is not _ObjDoesNotExist:
+                        result[action][path_str][elem] = elem_value
+            elif action == 'iterable_item_moved':
+                result[action][path_str] = {
+                    'new_path': stringify_path(
+                        flat_dict.get('new_path', ''),
+                        root_element=('root', GET)
+                    ),
+                    'value': value,
+                }
+
+        return result
+
     def to_flat_dicts(self, include_action_in_path=False, report_type_changes=True):
         """
         Returns a flat list of actions that is easily machine readable.

deepdiff/model.py

@@ -5,6 +5,7 @@
 from deepdiff.helper import (
     RemapDict, strings, short_repr, notpresent, get_type, numpy_numbers, np, literal_eval_extended,
     dict_)
+from deepdiff.path import stringify_element
 
 logger = logging.getLogger(__name__)
 
@@ -874,21 +875,7 @@ def stringify_param(self, force=None):
         """
         param = self.param
         if isinstance(param, strings):
-            has_quote = "'" in param
-            has_double_quote = '"' in param
-            if has_quote and has_double_quote:
-                new_param = []
-                for char in param:
-                    if char in {'"', "'"}:
-                        new_param.append('\\')
-                    new_param.append(char)
-                param = ''.join(new_param)
-            elif has_quote:
-                result = f'"{param}"'
-            elif has_double_quote:
-                result = f"'{param}'"
-            else:
-                result = param if self.quote_str is None else self.quote_str.format(param)
+            result = stringify_element(param, quote_str=self.quote_str)
         elif isinstance(param, tuple):  # Currently only for numpy ndarrays
             result = ']['.join(map(repr, param))
         else:

deepdiff/path.py

@@ -21,10 +21,17 @@ def _add_to_elements(elements, elem, inside):
     if not elem:
         return
     if not elem.startswith('__'):
-        try:
-            elem = literal_eval(elem)
-        except (ValueError, SyntaxError):
-            pass
+        remove_quotes = False
+        if '\\' in elem:
+            remove_quotes = True
+        else:
+            try:
+                elem = literal_eval(elem)
+                remove_quotes = False
+            except (ValueError, SyntaxError):
+                remove_quotes = True
+        if remove_quotes and elem[0] == elem[-1] and elem[0] in {'"', "'"}:
+            elem = elem[1: -1]
         action = GETATTR if inside == '.' else GET
         elements.append((elem, action))
 
@@ -229,3 +236,51 @@ def parse_path(path, root_element=DEFAULT_FIRST_ELEMENT, include_actions=False):
     if include_actions is False:
         return [i[0] for i in result]
     return [{'element': i[0], 'action': i[1]} for i in result]
+
+
+def stringify_element(param, quote_str=None):
+    has_quote = "'" in param
+    has_double_quote = '"' in param
+    if has_quote and has_double_quote:
+        new_param = []
+        for char in param:
+            if char in {'"', "'"}:
+                new_param.append('\\')
+            new_param.append(char)
+        param = ''.join(new_param)
+    elif has_quote:
+        result = f'"{param}"'
+    elif has_double_quote:
+        result = f"'{param}'"
+    else:
+        result = param if quote_str is None else quote_str.format(param)
+    return result
+
+
+def stringify_path(path, root_element=DEFAULT_FIRST_ELEMENT, quote_str="'{}'"):
+    """
+    Gets the path as an string.
+
+    For example [1, 2, 'age'] should become
+    root[1][2]['age']
+    """
+    if not path:
+        return root_element[0]
+    result = [root_element[0]]
+    has_actions = False
+    try:
+        if path[0][1] in {GET, GETATTR}:
+            has_actions = True
+    except (KeyError, IndexError, TypeError):
+        pass
+    if not has_actions:
+        path = [(i, GET) for i in path]
+        path[0] = (path[0][0], root_element[1])  # The action for the first element might be a GET or GETATTR. We update the action based on the root_element.
+    for element, action in path:
+        if isinstance(element, str) and action == GET:
+            element = stringify_element(element, quote_str)
+        if action == GET:
+            result.append(f"[{element}]")
+        else:
+            result.append(f".{element}")
+    return ''.join(result)