Feature: Dotted Set with Indexing (#1001)

* working set dot traversal
* working dotted set with index
* add unit test for setting envionrment variables using nested separators
* fixed issue 905
* change from double to triple underscore for indexing
* use triple underscore for INDEX_SEPARATOR_FOR_DYNACONF; assuming the separator is followed by an arbitrary number of digits as index

---------

Co-authored-by: sapphire008 <cui23327@gmail.com>
Co-authored-by: Bruno Rocha <rochacbruno@users.noreply.github.com>

docs/configuration.md

@@ -368,7 +368,7 @@ Otherwise it will be only what is specified in the latest loaded file. read more
 
 ---
 
-### **nested_separator**
+### **nested_separators**
 
 > type=`str`, default=`"__"` (double underescore) </br>
 > env-var=`NESTED_SEPARATOR_FOR_DYNACONF`
@@ -393,6 +393,34 @@ DATABASES = {
 }
 ```
 
+To access list by indexing
+
+> type=`str`, default=`"___"` (triple underescore) </br>
+> env-var=`INDEX_SEPARATOR_FOR_DYNACONF`
+
+ex:
+
+```bash
+export DYNACONF_DATABASES__default__WORKERS___0__Address="1.1.1.1"
+export DYNACONF_DATABASES__default__WORKERS___1__Address="2.2.2.2"
+```
+
+generates:
+
+```python
+DATABASES = {
+    "default": {
+        "ENGINE": {
+            "Address": "0.0.0.0"
+        },
+        "WORKERS": [
+            {"Address": "1.1.1.1"},
+            {"Address": "2.2.2.2"} 
+        ]
+    }
+}
+```
+
 !!! warning
     Choose something that is suitable for env vars, usually you don't need to change this variable.
 

dynaconf/base.py

@@ -4,6 +4,7 @@
 import importlib
 import inspect
 import os
+import re
 import warnings
 from collections import defaultdict
 from contextlib import contextmanager
@@ -887,15 +888,34 @@ def _dotted_set(
                 "VALIDATE_ON_UPDATE_FOR_DYNACONF"
             )  # pragma: nocover
 
-        split_keys = dotted_key.split(".")
+        # Add a "." before "[" to help splitting
+        split_keys = dotted_key.replace("[", ".[").split(".")
         existing_data = self.get(split_keys[0], {})
         new_data = tree = DynaBox(box_settings=self)
+        value = parse_conf_data(value, tomlfy=tomlfy, box_settings=self)
 
-        for k in split_keys[:-1]:
-            tree = tree.setdefault(k, {})
+        for n, k in enumerate(split_keys):
+            is_not_end = n < (len(split_keys) - 1)
+            if is_not_end:
+                next_default = [] if "[" in split_keys[n + 1] else {}
 
-        value = parse_conf_data(value, tomlfy=tomlfy, box_settings=self)
-        tree[split_keys[-1]] = value
+            if "[" not in k:  # accessing field of a dict
+                if is_not_end:
+                    tree = tree.setdefault(k, next_default)  # get next
+                else:
+                    tree[k] = value  # assign value
+            elif k.startswith("[") and k.endswith(
+                "]"
+            ):  # accessing index of a list
+                index = int(k.replace("[", "").replace("]", ""))
+                # This makes sure we can assign any arbitrary index
+                tree.extend([next_default] * (index + 1))
+                if is_not_end:
+                    tree = tree[index]  # get at index
+                else:
+                    tree[index] = value  # assign value
+            else:  # odd cases like [2]0
+                raise (ValueError("Invalid field:", k))
 
         if existing_data:
             old_data = DynaBox(
@@ -905,6 +925,7 @@ def _dotted_set(
                 old=old_data,
                 new=new_data,
                 full_path=split_keys,
+                list_merge="deep",  # when to use deep / shallow replace?
             )
         self.update(data=new_data, tomlfy=tomlfy, validate=validate, **kwargs)
 
@@ -945,13 +966,22 @@ def set(
             validate = self.get("VALIDATE_ON_UPDATE_FOR_DYNACONF")
         if dotted_lookup is empty:
             dotted_lookup = self.get("DOTTED_LOOKUP_FOR_DYNACONF")
+
+        # Do index replacement first
+        nested_ind = self.get("INDEX_SEPARATOR_FOR_DYNACONF")
+        # DYNACONF_DATA__a___0__key___2__subkey ->
+        # DYNACONF_DATA__a[0]__key[2]__subkey
+        if nested_ind and isinstance(key, str):
+            nested_ind = rf"{nested_ind}(\d+)"
+            key = re.sub(nested_ind, r"[\1]", key)
+
         nested_sep = self.get("NESTED_SEPARATOR_FOR_DYNACONF")
 
         if isinstance(key, str):
             if nested_sep and nested_sep in key:
                 key = key.replace(nested_sep, ".")  # FOO__bar -> FOO.bar
 
-            if "." in key and dotted_lookup is True:
+            if ("." in key or "[" in key) and dotted_lookup is True:
                 return self._dotted_set(
                     key,
                     value,
@@ -1003,7 +1033,15 @@ def set(
                 # `dynaconf_merge` may be used within the key structure
                 # Or merge_enabled is set to True
                 parsed, source_metadata = self._merge_before_set(
-                    existing, parsed, source_metadata, context_merge=merge
+                    existing,
+                    parsed,
+                    source_metadata,
+                    context_merge=merge,
+                    list_merge="replace"
+                    if source_metadata
+                    and source_metadata.loader == "setdefault"
+                    and source_metadata.env == "development"
+                    else "merge",  # fix 905
                 )
 
         if isinstance(parsed, dict) and not isinstance(parsed, DynaBox):
@@ -1070,6 +1108,7 @@ def update(
 
         data = data or {}
         data.update(kwargs)
+
         for key, value in data.items():
             # update() will handle validation later
             with suppress(ValidationError):
@@ -1095,6 +1134,7 @@ def _merge_before_set(
         value,
         identifier: SourceMetadata | None = None,
         context_merge=empty,
+        list_merge="merge",
     ):
         """
         Merge the new value being set with the existing value before set
@@ -1116,7 +1156,7 @@ def _merge_before_set(
                 identifier = (
                     identifier._replace(merged=True) if identifier else None
                 )
-                value = object_merge(existing, value)
+                value = object_merge(existing, value, list_merge=list_merge)
 
         if isinstance(value, (list, tuple)):
             value = list(value)

dynaconf/default_settings.py

@@ -144,6 +144,17 @@ def reload(load_dotenv=None, *args, **kwargs):
 # To disable it one can set `NESTED_SEPARATOR_FOR_DYNACONF=false`
 NESTED_SEPARATOR_FOR_DYNACONF = get("NESTED_SEPARATOR_FOR_DYNACONF", "__")
 
+# By default `__(\d+)` (regexp) is the separated for nested list env vars
+# export `DYNACONF_DATABASE_servers__0=server.com`
+# export `DYNACONF_DATABASE_servers__1=server.org`
+# export `DYNACONF_DATABASE_servers_2_0=server.io`
+# Should result in settings.DATABASE == {
+#   "servers": ["server.com", "server.org"],
+#   "servers_2_0": "server.io"
+# }
+# To disable it one can set `INDEX_SEPARATOR_FOR_DYNACONF=false`
+INDEX_SEPARATOR_FOR_DYNACONF = get("INDEX_SEPARATOR_FOR_DYNACONF", "___")
+
 # The env var specifying settings module
 ENVVAR_FOR_DYNACONF = get("ENVVAR_FOR_DYNACONF", "SETTINGS_FILE_FOR_DYNACONF")
 

dynaconf/utils/__init__.py

@@ -6,6 +6,7 @@
 from json import JSONDecoder
 from typing import Any
 from typing import Iterator
+from typing import Literal
 from typing import TYPE_CHECKING
 from typing import TypeVar
 
@@ -30,7 +31,11 @@
 
 
 def object_merge(
-    old: Any, new: Any, unique: bool = False, full_path: list[str] = None
+    old: Any,
+    new: Any,
+    unique: bool = False,
+    full_path: list[str] = None,
+    list_merge: Literal["merge", "shallow", "deep"] = "merge",
 ) -> Any:
     """
     Recursively merge two data structures, new is mutated in-place.
@@ -39,6 +44,11 @@ def object_merge(
     :param new: The new data to get old values merged in to.
     :param unique: When set to True existing list items are not set.
     :param full_path: Indicates the elements of a tree.
+    :param list_merge: Methods to use to merge lists
+        - merge: default merge behavior, i.e. (unique) concatenation
+        - shallow: replace the top-most level list of the nested structure
+        - deep: iteratively traverse the nested structure and replace
+            the element in the list at the level specified by the full_path
     """
     if full_path is None:
         full_path = []
@@ -52,10 +62,29 @@ def object_merge(
             new.remove("dynaconf_merge_unique")
             unique = True
 
-        for item in old[::-1]:
-            if unique and item in new:
-                continue
-            new.insert(0, item)
+        if list_merge == "merge" or unique:
+            for item in old[::-1]:
+                if unique and item in new:
+                    continue
+                new.insert(0, item)
+        # replace mode
+        # elif list_merge == "replace": pass
+        elif len(full_path) > 0:  # element-wise merge
+            new.extend([[]] * max(len(old) - len(new), 0))
+            for ii, item in enumerate(old):
+                # replace at corresponding positions
+                if list_merge == "shallow":
+                    new[ii] = new[ii] or item
+                else:  # deep replace
+                    if not new[ii]:  # copy over the older values
+                        new[ii] = item
+                    elif item:  # old[ii] is not None
+                        object_merge(
+                            old[ii],
+                            new[ii],
+                            full_path=full_path[1:],
+                            list_merge="deep",
+                        )
 
     if isinstance(old, dict) and isinstance(new, dict):
         existing_value = recursive_get(old, full_path)  # doesn't handle None
@@ -96,13 +125,13 @@ def safe_items(data):
                 if old_key not in new:
                     new[old_key] = value
                 else:
-                    object_merge(
+                    new[old_key] = object_merge(
                         value,
                         new[old_key],
                         full_path=full_path[1:] if full_path else None,
+                        list_merge=list_merge,
                     )
-
-        handle_metavalues(old, new)
+        handle_metavalues(old, new, list_merge=list_merge)
 
     return new
 
@@ -111,22 +140,39 @@ def recursive_get(
     obj: DynaBox | dict[str, int] | dict[str, str | int],
     names: list[str] | None,
 ) -> Any:
-    """Given a dot accessible object and a list of names `foo.bar.zaz`
-    gets recursively all names one by one obj.foo.bar.zaz.
+    """Given a dot accessible object and a list of names `foo.bar.[1].zaz`
+    gets recursively all names one by one obj.foo.bar.[1].zaz.
     """
-    if not names:
+    if not names or obj is None:
         return
     head, *tail = names
-    result = getattr(obj, head, None)
+    if "[" not in head:
+        result = getattr(obj, head, None)
+    else:
+        index = int(head.replace("[", "").replace("]", ""))
+        result = obj[index] if index < len(obj) else []
+
     if not tail:
         return result
+
     return recursive_get(result, tail)
 
 
 def handle_metavalues(
-    old: DynaBox | dict[str, int] | dict[str, str | int], new: Any
+    old: DynaBox | dict[str, int] | dict[str, str | int],
+    new: Any,
+    list_merge: Literal["merge", "shallow", "deep"] = "merge",
 ) -> None:
-    """Cleanup of MetaValues on new dict"""
+    """
+    Cleanup of MetaValues on new dict
+    :param old: old values
+    :param new: new values
+    :param list_merge: Methods to use to merge lists
+        - merge: default merge behavior, i.e. (unique) concatenation
+        - shallow: replace the top-most level list of the nested structure
+        - deep: iteratively traverse the nested structure and replace
+            the element in the list at the level specified by the full_path
+    """
 
     for key in list(new.keys()):
         # MetaValue instances
@@ -140,7 +186,9 @@ def handle_metavalues(
         elif getattr(new[key], "_dynaconf_merge", False):
             # a Merge on `new` triggers merge with existing data
             new[key] = object_merge(
-                old.get(key), new[key].unwrap(), unique=new[key].unique
+                old.get(key),
+                new[key].unwrap(),
+                unique=new[key].unique,
             )
 
         # Data structures containing merge tokens
@@ -173,7 +221,9 @@ def handle_metavalues(
                 new[key] = local_merge
 
             if local_merge:
-                new[key] = object_merge(old.get(key), new[key])
+                new[key] = object_merge(
+                    old.get(key), new[key], list_merge=list_merge
+                )
 
 
 class DynaconfDict(dict):

tests/test_base.py

@@ -700,6 +700,40 @@ def test_dotted_set(settings):
     }
 
 
+def test_dotted_set_with_indexing(settings):
+    settings.set("MERGE_ENABLED_FOR_DYNACONF", False)
+
+    # Dotted set with index
+    settings.set("nested_a.nested_b[2][1].nested_c.nested_d[3]", "old_conf")
+    settings.set(
+        "nested_a.nested_b[2][1].nested_c.nested_d[3]", "new_conf1"
+    )  # overwrite
+    settings.set(
+        "nested_a.nested_b[2][1].nested_c.nested_d[2]", "new_conf2"
+    )  # insert
+    assert settings.NESTED_A.NESTED_B[2][1].NESTED_C.NESTED_D[3] == "new_conf1"
+    assert settings.NESTED_A.NESTED_B[2][1].NESTED_C.NESTED_D[2] == "new_conf2"
+    assert len(settings.NESTED_A.NESTED_B[0]) < 1
+    settings.set(
+        "nested_a.nested_b[0][2].nested_c.nested_d[0]", "extra_conf"
+    )  # add more
+    assert len(settings.NESTED_A.NESTED_B[0]) > 0
+    settings.set(
+        "nested_a.nested_b[2][1].nested_c.nested_d",
+        ["conf1", "conf2", "conf3"],
+    )  # overwrite list
+    assert settings.NESTED_A.NESTED_B[2][1].NESTED_C.NESTED_D == [
+        "conf1",
+        "conf2",
+        "conf3",
+    ]
+
+    # This test case is the reason why choosing
+    # __(\d+) pattern instead of _(\d+)_
+    settings.set("nested_5.nested_6_0", "World")
+    assert settings.NESTED_5.NESTED_6_0 == "World"
+
+
 def test_dotted_set_with_merge(settings):
     settings.set("MERGE_ENABLED_FOR_DYNACONF", False)
 
@@ -1508,3 +1542,21 @@ def test_no_extra_values_in_nested_structure():
     settings = Dynaconf()
     settings.set("key", [{"d": "v"}])
     assert settings.key == [{"d": "v"}]
+
+
+def test_environ_dotted_set_with_index():
+    os.environ["DYNACONF_NESTED_A__nested_1__nested_2"] = "new_conf"
+    os.environ[
+        "DYNACONF_NESTED_A__nested_b___2___1__nested_c__nested_d___3"
+    ] = "old_conf"
+    settings = Dynaconf(envvar_prefix="DYANCONF")
+    assert isinstance(settings.NESTED_A.NESTED_B, list)
+    assert isinstance(settings.NESTED_A.NESTED_B[2], list)
+    assert isinstance(settings.NESTED_A.NESTED_B[2][1], dict)
+    assert settings.NESTED_A.NESTED_B[2][1].NESTED_C.NESTED_D[3] == "old_conf"
+    assert settings.NESTED_A.NESTED_1.NESTED_2 == "new_conf"
+    # remove environment variables after testing
+    del os.environ["DYNACONF_NESTED_A__nested_1__nested_2"]
+    del os.environ[
+        "DYNACONF_NESTED_A__nested_b___2___1__nested_c__nested_d___3"
+    ]