Nested env

changes/3159-Air-Mark.md

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

docs/examples/settings_nested_env.py

@@ -0,0 +1,41 @@
+import os
+
+from pydantic import BaseModel, BaseSettings
+
+
+class SubSubValue(BaseModel):
+    v6: str
+
+
+class SubValue(BaseModel):
+    v4: str
+    v5: str
+    sub_sub: SubSubValue
+
+
+class TopValue(BaseModel):
+    v1: str
+    v2: str
+    v3: str
+    sub: SubValue
+
+
+class Settings(BaseSettings):
+    v0: str
+    top: TopValue
+
+    class Config:
+        env_nested_delimiter = '__'
+
+
+os.environ.update({
+    'top': '{"v1": "1", "v2": "2"}',
+    'v0': '0',
+    'top__v3': '3',
+    'top__sub': '{"sub_sub": {"v6": "6"}}',
+    'top__sub__v4': '4',
+    'top__sub__v5': '5',
+})
+
+
+print(Settings())

docs/usage/settings.md

@@ -69,6 +69,31 @@ 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 TOP='{"v1": "1", "v2": "2"}'
+export V0=0
+export TOP__V3=3
+export TOP__SUB='{"sub_sub": {"v6": "6"}}'
+export TOP__SUB__V4=4
+export TOP__SUB__V5=5
+```
+
+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.
+
 ## 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,32 +142,24 @@ 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]:
         """
         Build environment variables suitable for passing to the Model.
         """
         d: Dict[str, Optional[str]] = {}
 
-        if settings.__config__.case_sensitive:
-            env_vars: Mapping[str, Optional[str]] = os.environ
-        else:
-            env_vars = {k.lower(): v for k, v in os.environ.items()}
-
-        if self.env_file is not None:
-            env_path = Path(self.env_file).expanduser()
-            if env_path.is_file():
-                env_vars = {
-                    **read_env_file(
-                        env_path, encoding=self.env_file_encoding, case_sensitive=settings.__config__.case_sensitive
-                    ),
-                    **env_vars,
-                }
+        env_vars = self.get_env_vars(settings)
+        if self.env_nested_delimiter is not None:
+            env_vars = self.explode_env_vars(settings, env_vars)
 
         for field in settings.__fields__.values():
             env_val: Optional[str] = None
@@ -170,25 +172,70 @@ def __call__(self, settings: BaseSettings) -> Dict[str, Any]:
                 continue
 
             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
+                if isinstance(env_val, (str, bytes, bytearray)):
+                    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_origin(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
+                if isinstance(env_val, (str, bytes, bytearray)):
+                    try:
+                        env_val = settings.__config__.json_loads(env_val)
+                    except ValueError:
+                        pass
             d[field.alias] = env_val
         return d
 
     def __repr__(self) -> str:
         return f'EnvSettingsSource(env_file={self.env_file!r}, env_file_encoding={self.env_file_encoding!r})'
 
+    def get_env_vars(self, settings: BaseSettings) -> Mapping[str, Optional[str]]:
+        if settings.__config__.case_sensitive:
+            env_vars: Mapping[str, Optional[str]] = os.environ
+        else:
+            env_vars = {k.lower(): v for k, v in os.environ.items()}
+
+        if self.env_file is not None:
+            env_path = Path(self.env_file).expanduser()
+            if env_path.is_file():
+                env_vars = {
+                    **read_env_file(
+                        env_path, encoding=self.env_file_encoding, case_sensitive=settings.__config__.case_sensitive
+                    ),
+                    **env_vars,
+                }
+        return env_vars
+
+    def explode_env_vars(self, settings: BaseSettings, env_vars: Mapping[str, Optional[str]]) -> Dict[str, Any]:
+        """
+        Go trough environment variables.
+        Within the each key-value pair:
+            - Split key with the env_nested_delimiter
+            - Before going through each key define env_var as a root (same as the final result)
+            - Go trough each key, add a new dict item to the env_var and redefine env_var within this item
+            - At the last iteration:
+                - Silently try to parse value with the json, leave it as it is if it fails
+                - Assign the value to the deepest key
+        """
+        result: Dict[str, Any] = {}
+        for env_name, env_val in env_vars.items():
+            keys = env_name.split(self.env_nested_delimiter)
+            env_var = result
+            for idx, key in enumerate(keys):
+                if idx == len(keys) - 1:
+                    try:
+                        env_val = settings.__config__.json_loads(env_val)  # type: ignore
+                    except (ValueError, TypeError):
+                        ...
+                    env_var[key] = env_val
+                else:
+                    env_var = env_var.setdefault(key, {})
+        return result
+
 
 class SecretsSettingsSource:
     __slots__ = ('secrets_dir',)

tests/test_settings.py

@@ -90,6 +90,46 @@ 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: str
+        sub_sub: SubSubValue
+
+    class TopValue(BaseSettings):
+        v1: str
+        v2: str
+        v3: str
+        sub: SubValue
+
+    class Cfg(BaseSettings):
+        v0: str
+        top: TopValue
+
+        class Config:
+            env_nested_delimiter = '__'
+
+    env.set('top', '{"v1": "1", "v2": "2", "sub": {"v5": "xx"}}')
+    env.set('v0', '0')
+    env.set('top__v3', '3')
+    env.set('top__sub', '{"sub_sub": {"v6": "6"}}')
+    env.set('top__sub__v4', '4')
+    env.set('top__sub__v5', '5')
+    cfg = Cfg()
+    assert cfg.dict() == {
+        'v0': '0',
+        'top': {
+            'v1': '1',
+            'v2': '2',
+            'v3': '3',
+            'sub': {'v4': '4', 'v5': '5', 'sub_sub': {'v6': '6'}},
+        },
+    }
+
+
 class DateModel(BaseModel):
     pips: bool = False