Merge 567df63 into 0971ac9

CHANGELOG.md

@@ -1,6 +1,7 @@
 # 0.9 (beta)
 
 - Fixed serialization of optional nested dataclasses with a value of `None`.
+- Fixed preservation of comments on nested dataclass attributes.
 
 # 0.8.1 (2020-03-30)
 

datafiles/hooks.py

@@ -3,6 +3,7 @@
 from functools import wraps
 
 import log
+from ruamel.yaml.comments import CommentedMap, CommentedSeq
 
 from . import settings
 from .mapper import create_mapper
@@ -27,14 +28,6 @@
 FLAG = '_patched'
 
 
-class List(list):
-    """Patchable `list` type."""
-
-
-class Dict(dict):
-    """Patchable `dict` type."""
-
-
 def apply(instance, mapper):
     """Path methods that get or set attributes."""
     cls = instance.__class__
@@ -56,10 +49,10 @@ def apply(instance, mapper):
         for attr_name in instance.datafile.attrs:
             attr = getattr(instance, attr_name)
             if isinstance(attr, list):
-                attr = List(attr)
+                attr = CommentedSeq(attr)
                 setattr(instance, attr_name, attr)
             elif isinstance(attr, dict):
-                attr = Dict(attr)
+                attr = CommentedMap(attr)
                 setattr(instance, attr_name, attr)
             elif not is_dataclass(attr):
                 continue

datafiles/mapper.py

@@ -138,12 +138,13 @@ def _get_data(self, include_default_values: Trilean = None) -> Dict:
 
             if getattr(converter, 'DATACLASS', None):
                 log.debug(f"Converting '{name}' dataclass with {converter}")
-                data[name] = converter.to_preserialization_data(
+                new_value = converter.to_preserialization_data(
                     value,
                     default_to_skip=Missing
                     if include_default_values
                     else get_default_field_value(self._instance, name),
                 )
+                data[name] = recursive_update(value, new_value)
 
             elif (
                 value == get_default_field_value(self._instance, name)
@@ -171,7 +172,7 @@ def _get_text(self, **kwargs) -> str:
 
     @text.setter  # type: ignore
     def text(self, value: str):
-        write(self.path, value.strip() + '\n')
+        write(self.path, value.strip() + '\n', display=True)
 
     def load(self, *, _log=True, _first_load=False) -> None:
         if self._root:
@@ -274,7 +275,7 @@ def save(self, *, include_default_values: Trilean = None, _log=True) -> None:
         with hooks.disabled():
             text = self._get_text(include_default_values=include_default_values)
 
-        write(self.path, text)
+        write(self.path, text, display=True)
 
         self.modified = False
 

datafiles/tests/test_utils.py

@@ -6,48 +6,119 @@
 
 
 def describe_recursive_update():
-    def it_preserves_root_id(expect):
-        old: Dict = {}
-        new = {'a': 1}
-        id_ = id(old)
+    def describe_id_preservation():
+        def with_dict(expect):
+            old = {'my_dict': {'a': 1}}
+            new = {'my_dict': {'a': 2}}
+            previous_id = id(old['my_dict'])
 
-        old = recursive_update(old, new)
+            recursive_update(old, new)
 
-        expect(old) == new
-        expect(id(old)) == id_
+            expect(old) == new
+            expect(id(old['my_dict'])) == previous_id
 
-    def it_preserves_nested_dict_id(expect):
-        old = {'a': {'b': 1}}
-        new = {'a': {'b': 2}}
-        id_ = id(old['a'])
+        def with_dict_value(expect):
+            old = {'my_dict': {'my_nested_dict': {'a': 1}}}
+            new = {'my_dict': {'my_nested_dict': {'a': 2}}}
+            previous_id = id(old['my_dict']['my_nested_dict'])
 
-        old = recursive_update(old, new)
+            recursive_update(old, new)
 
-        expect(old) == new
-        expect(id(old['a'])) == id_
+            expect(old) == new
+            expect(id(old['my_dict']['my_nested_dict'])) == previous_id
 
-    def it_preserves_nested_list_id(expect):
-        old = {'a': [1]}
-        new = {'a': [2]}
-        id_ = id(old['a'])
+        def with_list(expect):
+            old = {'my_list': [1]}
+            new = {'my_list': [2]}
+            previous_id = id(old['my_list'])
 
-        old = recursive_update(old, new)
+            recursive_update(old, new)
 
-        expect(old) == new
-        expect(id(old['a'])) == id_
+            expect(old) == new
+            expect(id(old['my_list'])) == previous_id
 
-    def it_adds_missing_dict(expect):
-        old: Dict = {}
-        new = {'a': {'b': 2}}
+        def with_list_item(expect):
+            old = {'my_list': [{'name': "John"}]}
+            new = {'my_list': [{'name': "Jane"}]}
+            previous_id = id(old['my_list'][0])
 
-        old = recursive_update(old, new)
+            recursive_update(old, new)
 
-        expect(old) == new
+            expect(old) == new
+            expect(id(old['my_list'][0])) == previous_id
 
-    def it_adds_missing_list(expect):
-        old: Dict = {}
-        new = {'a': [1]}
+        def with_nested_list(expect):
+            old = {'my_dict': {'my_list': [{'name': "John"}]}}
+            new = {'my_dict': {'my_list': [{'name': "Jane"}]}}
+            previous_id = id(old['my_dict']['my_list'][0])
 
-        old = recursive_update(old, new)
+            recursive_update(old, new)
 
-        expect(old) == new
+            expect(old) == new
+            expect(id(old['my_dict']['my_list'][0])) == previous_id
+
+        def with_unchanged_immutables(expect):
+            hello = "hello"
+            world = "world"
+
+            old = {'my_dict': {'my_str': hello + world}}
+            new = {'my_dict': {'my_str': "helloworld"}}
+            previous_id = id(old['my_dict']['my_str'])
+
+            recursive_update(old, new)
+
+            expect(old) == new
+            expect(id(old['my_dict']['my_str'])) == previous_id
+
+        def with_updated_types(expect):
+            old = {'x': 1}
+            new = {'x': 1.0}
+            previous_id = id(old['x'])
+
+            recursive_update(old, new)
+
+            expect(old) == new
+            expect(id(old['x'])) != previous_id
+
+    def describe_merge():
+        def with_shoreter_list_into_longer(expect):
+            old = {'my_list': [1, 2, 3]}
+            new = {'my_list': [5, 6]}
+
+            recursive_update(old, new)
+
+            expect(old) == new
+
+        def with_longer_list_into_shorter(expect):
+            old = {'my_list': [1, 2]}
+            new = {'my_list': [3, 4, 5]}
+
+            recursive_update(old, new)
+
+            expect(old) == new
+
+    def describe_missing():
+        def with_dict(expect):
+            old: Dict = {}
+            new = {'my_dict': {'a': 1}}
+
+            recursive_update(old, new)
+
+            expect(old) == new
+
+        def with_list(expect):
+            old: Dict = {}
+            new = {'my_list': [1]}
+
+            recursive_update(old, new)
+
+            expect(old) == new
+
+    def describe_extra():
+        def with_dict(expect):
+            old = {'my_dict': {'a': 1, 'b': 2}}
+            new = {'my_dict': {'a': 1}}
+
+            recursive_update(old, new)
+
+            expect(old) == new

datafiles/utils.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 from pprint import pformat
 from shutil import get_terminal_size
-from typing import Dict, Optional, Union
+from typing import Any, Dict, Optional, Union
 
 import log
 
@@ -52,24 +52,43 @@ def dictify(value):
     return value
 
 
-def recursive_update(old: Dict, new: Dict):
-    """Recursively update a dictionary."""
+def recursive_update(old: Dict, new: Dict) -> Dict:
+    """Recursively update a dictionary, keeping equivalent objects."""
+    return _merge(old, new)
 
-    for key, value in new.items():
-        if isinstance(value, dict):
-            if key in old:
-                recursive_update(old[key], value)
-            else:
-                old[key] = value
-        elif isinstance(value, list):
-            if key in old:
-                old[key][:] = value
-            else:
-                old[key] = value
-        else:
-            old[key] = value
 
-    return old
+def _merge(old: Any, new: Any) -> Any:
+    if old is None:
+        return new
+
+    if isinstance(new, dict):
+        for key, value in new.items():
+            old[key] = _merge(old.get(key), value)
+
+        for key in list(old.keys()):
+            if key not in new:
+                old.pop(key)
+
+        return old
+
+    if isinstance(new, list):
+        for index, new_item in enumerate(new):
+            try:
+                old_item = old[index]
+            except IndexError:
+                old_item = None
+                old.append(old_item)
+            old[index] = _merge(old_item, new_item)
+
+        while len(old) > len(new):
+            old.pop()
+
+        return old
+
+    if new == old and isinstance(new, type(old)):
+        return old
+
+    return new
 
 
 def dedent(text: str) -> str:
@@ -79,8 +98,8 @@ def dedent(text: str) -> str:
     return text.replace('    ' * indent, '')
 
 
-def write(filename_or_path: Union[str, Path], text: str) -> None:
-    """Write text to a given file with logging."""
+def write(filename_or_path: Union[str, Path], text: str, *, display=False) -> None:
+    """Write text to a given file and optionally log it."""
     if isinstance(filename_or_path, Path):
         path = filename_or_path
     else:
@@ -93,14 +112,17 @@ def write(filename_or_path: Union[str, Path], text: str) -> None:
         content = text.replace(' \n', '␠\n')
     else:
         content = '∅\n'
-    log.debug(message + '\n' + line + '\n' + content + line)
+    if display:
+        log.debug(message + '\n' + line + '\n' + content + line)
+    else:
+        log.critical(message)
 
     path.parent.mkdir(parents=True, exist_ok=True)
     path.write_text(text)
 
 
-def read(filename: str) -> str:
-    """Read text from a file with logging."""
+def read(filename: str, *, display=False) -> str:
+    """Read text from a file and optionally log it."""
     path = Path(filename).resolve()
     message = f'Reading file: {path}'
     line = '=' * (31 + len(message))
@@ -109,7 +131,10 @@ def read(filename: str) -> str:
         content = text.replace(' \n', '␠\n')
     else:
         content = '∅\n'
-    log.debug(message + '\n' + line + '\n' + content + line)
+    if display:
+        log.debug(message + '\n' + line + '\n' + content + line)
+    else:
+        log.critical(message)
     return text