dictutils: adds utils for dot lookup and clear

* Adds a method "clear_none()" which helps removing None and empty
  list/dict values from the dictionary.

* Adds a method "dict_lookup()" to support dot key string notation for
  looking up a key in a dictionary.

invenio_records/api.py

@@ -18,6 +18,7 @@
 from sqlalchemy_continuum.utils import parent_class
 from werkzeug.local import LocalProxy
 
+from .dictutils import clear_none, dict_lookup
 from .dumpers import Dumper
 from .errors import MissingModelError
 from .models import RecordMetadata
@@ -172,6 +173,13 @@ def replace_refs(self):
         """Replace the ``$ref`` keys within the JSON."""
         return _records_state.replace_refs(self)
 
+    def clear_none(self, key=None):
+        """Helper method to clear None, empty dict and list values.
+
+        Modifications are done in place.
+        """
+        clear_none(dict_lookup(self, key) if key else self)
+
     def dumps(self, dumper=None):
         """Make a dump of the record (defaults to a deep copy of the dict).
 

invenio_records/dictutils.py

@@ -0,0 +1,98 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of Invenio.
+# Copyright (C) 2015-2020 CERN.
+#
+# Invenio is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Dictionary utilities."""
+
+from copy import deepcopy
+
+
+def clear_none(d):
+    """Clear None values and empty dicts from a dict."""
+    del_keys = []
+    for k, v in d.items():
+        if v is None:
+            del_keys.append(k)
+        elif isinstance(v, dict):
+            clear_none(v)
+            if v == {}:
+                del_keys.append(k)
+        elif isinstance(v, list):
+            clear_none_list(v)
+            if v == []:
+                del_keys.append(k)
+
+    # Delete the keys (cannot be done during the dict iteration)
+    for k in del_keys:
+        del d[k]
+
+
+def clear_none_list(ls):
+    """Clear values from a list (in-place)."""
+    del_idx = []
+    for i, v in enumerate(ls):
+        if v is None:
+            del_idx.append(i)
+        elif isinstance(v, list):
+            clear_none_list(v)
+            if v == []:
+                del_idx.append(i)
+        elif isinstance(v, dict):
+            clear_none(v)
+            if v == {}:
+                del_idx.append(i)
+
+    # Delete the keys (reverse so index stays stable).
+    for i in reversed(del_idx):
+        del ls[i]
+
+
+def dict_lookup(source, lookup_key, parent=False):
+    """Make a lookup into a dict based on a dot notation.
+
+    Examples of the supported dot notation:
+
+    - ``'a'`` - Equivalent to ``source['a']``
+    - ``'a.b'`` - Equivalent to ``source['a']['b']``
+    - ``'a.b.0'`` - Equivalent to ``source['a']['b'][0]`` (for lists)
+
+    List notation is also supported:
+
+    - `['a']``
+    - ``['a','b']``
+    - ``['a','b', 0]``
+
+    :param source: The dictionary object to perform the lookup in.
+    :param parent: If parent argument is True, returns the parent node of
+                   matched object.
+    :param lookup_key: A string using dot notation, or a list of keys.
+    """
+    # Copied from dictdiffer (CERN contributed part) and slightly modified.
+    if not lookup_key:
+        raise KeyError("No lookup key specified")
+
+    # Parse the list of keys
+    if isinstance(lookup_key, str):
+        keys = lookup_key.split('.')
+    elif isinstance(lookup_key, list):
+        keys = lookup_key
+    else:
+        raise TypeError('lookup must be string or list')
+
+    if parent:
+        keys = keys[:-1]
+
+    # Lookup the key
+    value = source
+    for key in keys:
+        try:
+            if isinstance(value, list):
+                key = int(key)
+            value = value[key]
+        except (TypeError, IndexError) as exc:
+            raise KeyError(lookup_key) from exc
+    return value

invenio_records/systemfields/constant.py

@@ -8,6 +8,7 @@
 
 """Constant system field."""
 
+from ..dictutils import dict_lookup
 from .base import SystemField
 
 
@@ -17,16 +18,19 @@ class ConstantField(SystemField):
     def __init__(self, key, value):
         """Initialize the field.
 
-        :param key: The key to set in the dictionary (only top-level keys
-                    supported currently).
+        :param key: The key to set in the dictionary (dot notation supported
+                    for nested lookup).
         :param value: The value to set for the key.
         """
         self.key = key
         self.value = value
 
     def pre_init(self, record, data, model=None):
         """Sets the key in the record during record instantiation."""
-        if self.key not in data:
+        try:
+            dict_lookup(data, self.key)
+        except KeyError:
+            # Key is not present, so add it.
             data[self.key] = self.value
 
     def __get__(self, instance, class_):
@@ -35,6 +39,7 @@ def __get__(self, instance, class_):
         if instance is None:
             return self
         # Instance access
-        if self.key in instance:
-            return instance[self.key]
-        return None
+        try:
+            return dict_lookup(instance, self.key)
+        except KeyError:
+            return None

run-tests.sh

@@ -2,11 +2,14 @@
 # -*- coding: utf-8 -*-
 #
 # This file is part of Invenio.
-# Copyright (C) 2015-2018 CERN.
+# Copyright (C) 2015-2020 CERN.
 #
 # Invenio is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
+# Usage:
+#   env DB=postgresql ./run-tests.sh
+
 python -m pydocstyle invenio_records tests docs && \
 python -m isort invenio_records tests --check-only --diff && \
 python -m check_manifest --ignore ".travis-*" && \

setup.py

@@ -52,7 +52,7 @@
 
 install_requires = [
     'arrow>=0.16.0',
-    'invenio-base>=1.2.0',
+    'invenio-base>=1.2.2',
     'invenio-celery>=1.2.0',
     'invenio-i18n>=1.2.0',
     'jsonpatch>=1.26',

tests/test_api.py

@@ -474,3 +474,10 @@ def test_reversed_works_for_revisions(testapp, database):
     reversed_revisions[0].revision_id == 3
     reversed_revisions[1].revision_id == 1
     reversed_revisions[2].revision_id == 0
+
+
+def test_clear_none(testapp, db):
+    """Test clear_none."""
+    record = Record({'a': None})
+    record.clear_none()
+    assert record == {}

tests/test_dictutils.py

@@ -0,0 +1,65 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of Invenio.
+# Copyright (C) 2020 CERN.
+#
+# Invenio is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""Test of dictionary utilities."""
+
+from copy import deepcopy
+
+import pytest
+
+from invenio_records.dictutils import clear_none, dict_lookup
+
+
+def test_clear_none():
+    """Test clearning of the dictionary."""
+    d = {
+        'a': None,
+        'b': {
+            'c': None
+        },
+        'd': ['1', None, []],
+        'e': [{'a': None, 'b': []}],
+    }
+
+    clear_none(d)
+    # Modifications are done in place, so gotta test after the function call.
+    assert d == {'d': ['1']}
+
+    d = {
+        'a': None,
+        'b': [
+            {'a': '1', 'b': None},
+        ],
+    }
+    clear_none(d)
+    # Modifications are done in place, so gotta test after the function call.
+    assert d == {'b': [{'a': '1'}]}
+
+
+def test_dict_lookup():
+    """Test lookup by a key."""
+    d = {
+        'a': 1,
+        'b': {
+            'c': None
+        },
+        'd': ['1', '2'],
+    }
+    assert dict_lookup(d, 'a') == d['a']
+    assert dict_lookup(d, 'b') == d['b']
+    assert dict_lookup(d, 'b.c') == d['b']['c']
+    assert dict_lookup(d, 'd') == d['d']
+    assert dict_lookup(d, 'd.0') == d['d'][0]
+    assert dict_lookup(d, 'd.1') == d['d'][1]
+    assert dict_lookup(d, 'd.-1') == d['d'][-1]
+
+    assert pytest.raises(KeyError, dict_lookup, d, 'x')
+    assert pytest.raises(KeyError, dict_lookup, d, 'a.x')
+    assert pytest.raises(KeyError, dict_lookup, d, 'b.x')
+    assert pytest.raises(KeyError, dict_lookup, d, 'b.c.0')
+    assert pytest.raises(KeyError, dict_lookup, d, 'd.3')