Nested env

changes/3159-Air-Mark.md

@@ -0,0 +1 @@
+Nested env variables can now be configured regarding the `env_nested_delimiter`

docs/build/exec_examples.py

@@ -148,6 +148,11 @@ def exec_examples():
         'my_auth_key': 'xxx',
         'my_api_key': 'xxx',
         'database_dsn': 'postgres://postgres@localhost:5432/env_db',
+        'v0': '0',
+        'sub_model': '{"v1": "json-1", "v2": "json-2"}',
+        'sub_model__v2': 'nested-2',
+        'sub_model__v3': '3',
+        'sub_model__deep__v4': 'v4',
     })
 
     sys.path.append(str(EXAMPLES_DIR))

docs/examples/settings_nested_env.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel, BaseSettings
+
+
+class DeepSubModel(BaseModel):
+    v4: str
+
+
+class SubModel(BaseModel):
+    v1: str
+    v2: bytes
+    v3: int
+    deep: DeepSubModel
+
+
+class Settings(BaseSettings):
+    v0: str
+    sub_model: SubModel
+
+    class Config:
+        env_nested_delimiter = '__'
+
+
+print(Settings().dict())

docs/usage/settings.md

@@ -69,6 +69,36 @@ be if passed directly to the initialiser (as a string).
 Complex types like `list`, `set`, `dict`, and sub-models are populated from the environment
 by treating the environment variable's value as a JSON-encoded string.
 
+Another way to populate nested complex variables is to configure your model with the `env_nested_delimiter`
+config setting, then use an env variable with a name pointing to the nested module fields.
+What it does is simply explodes yor variable into nested models or dicts.
+So if you define a variable `FOO__BAR__BAZ=123` it will convert it into `FOO={'BAR': {'BAZ': 123}}`
+If you have multiple variables with the same structure they will be merged.
+
+With the following environment variables:
+```bash
+# your environment
+export V0=0
+export SUB_MODEL='{"v1": "json-1", "v2": "json-2"}'
+export SUB_MODEL__V2=nested-2
+export SUB_MODEL__V3=3
+export SUB_MODEL__DEEP__V4=v4
+```
+
+You could load a settings module thus:
+```py
+{!.tmp_examples/settings_nested_env.py!}
+```
+
+`env_nested_delimiter` can be configured via the `Config` class as shown above, or via the 
+`_env_nested_delimiter` keyword argument on instantiation.
+
+JSON is only parsed in top-level fields, if you need to parse JSON in sub-models, you will need to implement
+validators on those models.
+
+Nested environment variables take precedence over the top-level environment variable JSON
+(e.g. in the example above, `SUB_MODEL__V2` trumps `SUB_MODEL`).
+
 ## Dotenv (.env) support
 
 !!! note

pydantic/env_settings.py

@@ -30,13 +30,18 @@ def __init__(
         __pydantic_self__,
         _env_file: Optional[StrPath] = env_file_sentinel,
         _env_file_encoding: Optional[str] = None,
+        _env_nested_delimiter: Optional[str] = None,
         _secrets_dir: Optional[StrPath] = None,
         **values: Any,
     ) -> None:
         # Uses something other than `self` the first arg to allow "self" as a settable attribute
         super().__init__(
             **__pydantic_self__._build_values(
-                values, _env_file=_env_file, _env_file_encoding=_env_file_encoding, _secrets_dir=_secrets_dir
+                values,
+                _env_file=_env_file,
+                _env_file_encoding=_env_file_encoding,
+                _env_nested_delimiter=_env_nested_delimiter,
+                _secrets_dir=_secrets_dir,
             )
         )
 
@@ -45,6 +50,7 @@ def _build_values(
         init_kwargs: Dict[str, Any],
         _env_file: Optional[StrPath] = None,
         _env_file_encoding: Optional[str] = None,
+        _env_nested_delimiter: Optional[str] = None,
         _secrets_dir: Optional[StrPath] = None,
     ) -> Dict[str, Any]:
         # Configure built-in sources
@@ -54,6 +60,9 @@ def _build_values(
             env_file_encoding=(
                 _env_file_encoding if _env_file_encoding is not None else self.__config__.env_file_encoding
             ),
+            env_nested_delimiter=(
+                _env_nested_delimiter if _env_nested_delimiter is not None else self.__config__.env_nested_delimiter
+            ),
         )
         file_secret_settings = SecretsSettingsSource(secrets_dir=_secrets_dir or self.__config__.secrets_dir)
         # Provide a hook to set built-in sources priority and add / remove sources
@@ -71,6 +80,7 @@ class Config(BaseConfig):
         env_prefix = ''
         env_file = None
         env_file_encoding = None
+        env_nested_delimiter = None
         secrets_dir = None
         validate_all = True
         extra = Extra.forbid
@@ -132,17 +142,20 @@ def __repr__(self) -> str:
 
 
 class EnvSettingsSource:
-    __slots__ = ('env_file', 'env_file_encoding')
+    __slots__ = ('env_file', 'env_file_encoding', 'env_nested_delimiter')
 
-    def __init__(self, env_file: Optional[StrPath], env_file_encoding: Optional[str]):
+    def __init__(
+        self, env_file: Optional[StrPath], env_file_encoding: Optional[str], env_nested_delimiter: Optional[str] = None
+    ):
         self.env_file: Optional[StrPath] = env_file
         self.env_file_encoding: Optional[str] = env_file_encoding
+        self.env_nested_delimiter: Optional[str] = env_nested_delimiter
 
-    def __call__(self, settings: BaseSettings) -> Dict[str, Any]:
+    def __call__(self, settings: BaseSettings) -> Dict[str, Any]:  # noqa C901
         """
         Build environment variables suitable for passing to the Model.
         """
-        d: Dict[str, Optional[str]] = {}
+        d: Dict[str, Any] = {}
 
         if settings.__config__.case_sensitive:
             env_vars: Mapping[str, Optional[str]] = os.environ
@@ -166,26 +179,68 @@ def __call__(self, settings: BaseSettings) -> Dict[str, Any]:
                 if env_val is not None:
                     break
 
-            if env_val is None:
-                continue
+            is_complex, allow_json_failure = self.field_is_complex(field)
+            if is_complex:
+                if env_val is None:
+                    # field is complex but no value found so far, try explode_env_vars
+                    env_val_built = self.explode_env_vars(field, env_vars)
+                    if env_val_built:
+                        d[field.alias] = env_val_built
+                else:
+                    # field is complex and there's a value, decode that as JSON, then add explode_env_vars
+                    try:
+                        env_val = settings.__config__.json_loads(env_val)
+                    except ValueError as e:
+                        if not allow_json_failure:
+                            raise SettingsError(f'error parsing JSON for "{env_name}"') from e
+
+                    if isinstance(env_val, dict):
+                        d[field.alias] = deep_update(env_val, self.explode_env_vars(field, env_vars))
+                    else:
+                        d[field.alias] = env_val
+            elif env_val is not None:
+                # simplest case, field is not complex, we only need to add the value if it was found
+                d[field.alias] = env_val
 
-            if field.is_complex():
-                try:
-                    env_val = settings.__config__.json_loads(env_val)
-                except ValueError as e:
-                    raise SettingsError(f'error parsing JSON for "{env_name}"') from e
-            elif (
-                is_union(get_origin(field.type_)) and field.sub_fields and any(f.is_complex() for f in field.sub_fields)
-            ):
-                try:
-                    env_val = settings.__config__.json_loads(env_val)
-                except ValueError:
-                    pass
-            d[field.alias] = env_val
         return d
 
+    def field_is_complex(self, field: ModelField) -> Tuple[bool, bool]:
+        """
+        Find out if a field is complex, and if so whether JSON errors should be ignored
+        """
+        if field.is_complex():
+            allow_json_failure = False
+        elif is_union(get_origin(field.type_)) and field.sub_fields and any(f.is_complex() for f in field.sub_fields):
+            allow_json_failure = True
+        else:
+            return False, False
+
+        return True, allow_json_failure
+
+    def explode_env_vars(self, field: ModelField, env_vars: Mapping[str, Optional[str]]) -> Dict[str, Any]:
+        """
+        Process env_vars and extract the values of keys containing env_nested_delimiter into nested dictionaries.
+
+        This is applied to a single field, hence filtering by env_var prefix.
+        """
+        prefixes = [f'{env_name}{self.env_nested_delimiter}' for env_name in field.field_info.extra['env_names']]
+        result: Dict[str, Any] = {}
+        for env_name, env_val in env_vars.items():
+            if not any(env_name.startswith(prefix) for prefix in prefixes):
+                continue
+            _, *keys, last_key = env_name.split(self.env_nested_delimiter)
+            env_var = result
+            for key in keys:
+                env_var = env_var.setdefault(key, {})
+            env_var[last_key] = env_val
+
+        return result
+
     def __repr__(self) -> str:
-        return f'EnvSettingsSource(env_file={self.env_file!r}, env_file_encoding={self.env_file_encoding!r})'
+        return (
+            f'EnvSettingsSource(env_file={self.env_file!r}, env_file_encoding={self.env_file_encoding!r}, '
+            f'env_nested_delimiter={self.env_nested_delimiter!r})'
+        )
 
 
 class SecretsSettingsSource:

tests/test_settings.py

@@ -79,7 +79,7 @@ class Settings(BaseSettings):
     assert s.top == {'apple': 'value', 'banana': 'secret_value'}
 
 
-def test_nested_env_with_dict(env):
+def test_merge_dict(env):
     class Settings(BaseSettings):
         top: Dict[str, str]
 
@@ -90,6 +90,80 @@ class Settings(BaseSettings):
     assert s.top == {'apple': 'value', 'banana': 'secret_value'}
 
 
+def test_nested_env_delimiter(env):
+    class SubSubValue(BaseSettings):
+        v6: str
+
+    class SubValue(BaseSettings):
+        v4: str
+        v5: int
+        sub_sub: SubSubValue
+
+    class TopValue(BaseSettings):
+        v1: str
+        v2: str
+        v3: str
+        sub: SubValue
+
+    class Cfg(BaseSettings):
+        v0: str
+        v0_union: Union[SubValue, int]
+        top: TopValue
+
+        class Config:
+            env_nested_delimiter = '__'
+
+    env.set('top', '{"v1": "json-1", "v2": "json-2", "sub": {"v5": "xx"}}')
+    env.set('top__sub__v5', '5')
+    env.set('v0', '0')
+    env.set('top__v2', '2')
+    env.set('top__v3', '3')
+    env.set('v0_union', '0')
+    env.set('top__sub__sub_sub__v6', '6')
+    env.set('top__sub__v4', '4')
+    cfg = Cfg()
+    assert cfg.dict() == {
+        'v0': '0',
+        'v0_union': 0,
+        'top': {
+            'v1': 'json-1',
+            'v2': '2',
+            'v3': '3',
+            'sub': {'v4': '4', 'v5': 5, 'sub_sub': {'v6': '6'}},
+        },
+    }
+
+
+def test_nested_env_delimiter_complex_required(env):
+    class Cfg(BaseSettings):
+        v: str = 'default'
+
+        class Config:
+            env_nested_delimiter = '__'
+
+    env.set('v__x', 'x')
+    env.set('v__y', 'y')
+    cfg = Cfg()
+    assert cfg.dict() == {'v': 'default'}
+
+
+def test_nested_env_delimiter_aliases(env):
+    class SubModel(BaseSettings):
+        v1: str
+        v2: str
+
+    class Cfg(BaseSettings):
+        sub_model: SubModel
+
+        class Config:
+            fields = {'sub_model': {'env': ['foo', 'bar']}}
+            env_nested_delimiter = '__'
+
+    env.set('foo__v1', '-1-')
+    env.set('bar__v2', '-2-')
+    assert Cfg().dict() == {'sub_model': {'v1': '-1-', 'v2': '-2-'}}
+
+
 class DateModel(BaseModel):
     pips: bool = False
 
@@ -1010,6 +1084,6 @@ def test_builtins_settings_source_repr():
     )
     assert (
         repr(EnvSettingsSource(env_file='.env', env_file_encoding='utf-8'))
-        == "EnvSettingsSource(env_file='.env', env_file_encoding='utf-8')"
+        == "EnvSettingsSource(env_file='.env', env_file_encoding='utf-8', env_nested_delimiter=None)"
     )
     assert repr(SecretsSettingsSource(secrets_dir='/secrets')) == "SecretsSettingsSource(secrets_dir='/secrets')"