add deep merge functionality to config file sources (#698)

Author: pmeier

docs/index.md

@@ -2401,12 +2401,7 @@ Other settings sources are available for common configuration files:
 - `TomlConfigSettingsSource` using `toml_file` argument
 - `YamlConfigSettingsSource` using `yaml_file` and yaml_file_encoding arguments
 
-You can also provide multiple files by providing a list of path:
-```py
-toml_file = ['config.default.toml', 'config.custom.toml']
-```
-To use them, you can use the same mechanism described [here](#customise-settings-sources)
-
+To use them, you can use the same mechanism described [here](#customise-settings-sources).
 
 ```py
 from pydantic import BaseModel
@@ -2448,6 +2443,136 @@ foobar = "Hello"
 nested_field = "world!"
 ```
 
+You can also provide multiple files by providing a list of paths.
+
+```py
+from pydantic import BaseModel
+
+from pydantic_settings import (
+    BaseSettings,
+    PydanticBaseSettingsSource,
+    SettingsConfigDict,
+    TomlConfigSettingsSource,
+)
+
+
+class Nested(BaseModel):
+    foo: int
+    bar: int = 0
+
+
+class Settings(BaseSettings):
+    hello: str
+    nested: Nested
+    model_config = SettingsConfigDict(
+        toml_file=['config.default.toml', 'config.custom.toml']
+    )
+
+    @classmethod
+    def settings_customise_sources(
+        cls,
+        settings_cls: type[BaseSettings],
+        init_settings: PydanticBaseSettingsSource,
+        env_settings: PydanticBaseSettingsSource,
+        dotenv_settings: PydanticBaseSettingsSource,
+        file_secret_settings: PydanticBaseSettingsSource,
+    ) -> tuple[PydanticBaseSettingsSource, ...]:
+        return (TomlConfigSettingsSource(settings_cls),)
+```
+
+The following two configuration files
+
+```toml
+# config.default.toml
+hello = "World"
+
+[nested]
+foo = 1
+bar = 2
+```
+
+```toml
+# config.custom.toml
+[nested]
+foo = 3
+```
+
+are equivalent to
+
+```toml
+hello = "world"
+
+[nested]
+foo = 3
+```
+
+The files are merged shallowly in increasing order of priority. To enable deep merging, set `deep_merge=True` on the source directly.
+
+!!! warning
+    The `deep_merge` option is **not available** through the `SettingsConfigDict`.
+
+```py
+from pydantic import BaseModel
+
+from pydantic_settings import (
+    BaseSettings,
+    PydanticBaseSettingsSource,
+    SettingsConfigDict,
+    TomlConfigSettingsSource,
+)
+
+
+class Nested(BaseModel):
+    foo: int
+    bar: int = 0
+
+
+class Settings(BaseSettings):
+    hello: str
+    nested: Nested
+    model_config = SettingsConfigDict(
+        toml_file=['config.default.toml', 'config.custom.toml']
+    )
+
+    @classmethod
+    def settings_customise_sources(
+        cls,
+        settings_cls: type[BaseSettings],
+        init_settings: PydanticBaseSettingsSource,
+        env_settings: PydanticBaseSettingsSource,
+        dotenv_settings: PydanticBaseSettingsSource,
+        file_secret_settings: PydanticBaseSettingsSource,
+    ) -> tuple[PydanticBaseSettingsSource, ...]:
+        return (TomlConfigSettingsSource(settings_cls, deep_merge=True),)
+```
+
+With deep merge enabled, the following two configuration files
+
+```toml
+# config.default.toml
+hello = "World"
+
+[nested]
+foo = 1
+bar = 2
+```
+
+```toml
+# config.custom.toml
+[nested]
+foo = 3
+```
+
+are equivalent to
+
+```toml
+hello = "world"
+
+[nested]
+foo = 3
+bar = 2
+```
+
 ### pyproject.toml
 
 "pyproject.toml" is a standardized file for providing configuration values in Python projects.

pydantic_settings/sources/base.py

@@ -13,7 +13,7 @@
 from pydantic._internal._typing_extra import (  # type: ignore[attr-defined]
     get_origin,
 )
-from pydantic._internal._utils import is_model_class
+from pydantic._internal._utils import deep_update, is_model_class
 from pydantic.fields import FieldInfo
 from typing_inspection import typing_objects
 from typing_inspection.introspection import is_union_origin
@@ -192,16 +192,22 @@ def __call__(self) -> dict[str, Any]:
 
 
 class ConfigFileSourceMixin(ABC):
-    def _read_files(self, files: PathType | None) -> dict[str, Any]:
+    def _read_files(self, files: PathType | None, deep_merge: bool = False) -> dict[str, Any]:
         if files is None:
             return {}
         if isinstance(files, (str, os.PathLike)):
             files = [files]
         vars: dict[str, Any] = {}
         for file in files:
             file_path = Path(file).expanduser()
-            if file_path.is_file():
-                vars.update(self._read_file(file_path))
+            if not file_path.is_file():
+                continue
+
+            updating_vars = self._read_file(file_path)
+            if deep_merge:
+                vars = deep_update(vars, updating_vars)
+            else:
+                vars.update(updating_vars)
         return vars
 
     @abstractmethod

pydantic_settings/sources/providers/json.py

@@ -26,14 +26,15 @@ def __init__(
         settings_cls: type[BaseSettings],
         json_file: PathType | None = DEFAULT_PATH,
         json_file_encoding: str | None = None,
+        deep_merge: bool = False,
     ):
         self.json_file_path = json_file if json_file != DEFAULT_PATH else settings_cls.model_config.get('json_file')
         self.json_file_encoding = (
             json_file_encoding
             if json_file_encoding is not None
             else settings_cls.model_config.get('json_file_encoding')
         )
-        self.json_data = self._read_files(self.json_file_path)
+        self.json_data = self._read_files(self.json_file_path, deep_merge=deep_merge)
         super().__init__(settings_cls, self.json_data)
 
     def _read_file(self, file_path: Path) -> dict[str, Any]:

pydantic_settings/sources/providers/toml.py

@@ -50,9 +50,10 @@ def __init__(
         self,
         settings_cls: type[BaseSettings],
         toml_file: PathType | None = DEFAULT_PATH,
+        deep_merge: bool = False,
     ):
         self.toml_file_path = toml_file if toml_file != DEFAULT_PATH else settings_cls.model_config.get('toml_file')
-        self.toml_data = self._read_files(self.toml_file_path)
+        self.toml_data = self._read_files(self.toml_file_path, deep_merge=deep_merge)
         super().__init__(settings_cls, self.toml_data)
 
     def _read_file(self, file_path: Path) -> dict[str, Any]:

pydantic_settings/sources/providers/yaml.py

@@ -40,6 +40,7 @@ def __init__(
         yaml_file: PathType | None = DEFAULT_PATH,
         yaml_file_encoding: str | None = None,
         yaml_config_section: str | None = None,
+        deep_merge: bool = False,
     ):
         self.yaml_file_path = yaml_file if yaml_file != DEFAULT_PATH else settings_cls.model_config.get('yaml_file')
         self.yaml_file_encoding = (
@@ -52,7 +53,7 @@ def __init__(
             if yaml_config_section is not None
             else settings_cls.model_config.get('yaml_config_section')
         )
-        self.yaml_data = self._read_files(self.yaml_file_path)
+        self.yaml_data = self._read_files(self.yaml_file_path, deep_merge=deep_merge)
 
         if self.yaml_config_section:
             try:

tests/test_source_json.py

@@ -5,6 +5,7 @@
 import json
 from pathlib import Path
 
+import pytest
 from pydantic import BaseModel
 
 from pydantic_settings import (
@@ -98,3 +99,36 @@ def settings_customise_sources(
 
     s = Settings()
     assert s.model_dump() == {'json5': 5, 'json6': 6}
+
+
+@pytest.mark.parametrize('deep_merge', [False, True])
+def test_multiple_file_json_merge(tmp_path, deep_merge):
+    p5 = tmp_path / '.env.json5'
+    p6 = tmp_path / '.env.json6'
+
+    with open(p5, 'w') as f5:
+        json.dump({'hello': 'world', 'nested': {'foo': 1, 'bar': 2}}, f5)
+    with open(p6, 'w') as f6:
+        json.dump({'nested': {'foo': 3}}, f6)
+
+    class Nested(BaseModel):
+        foo: int
+        bar: int = 0
+
+    class Settings(BaseSettings):
+        hello: str
+        nested: Nested
+
+        @classmethod
+        def settings_customise_sources(
+            cls,
+            settings_cls: type[BaseSettings],
+            init_settings: PydanticBaseSettingsSource,
+            env_settings: PydanticBaseSettingsSource,
+            dotenv_settings: PydanticBaseSettingsSource,
+            file_secret_settings: PydanticBaseSettingsSource,
+        ) -> tuple[PydanticBaseSettingsSource, ...]:
+            return (JsonConfigSettingsSource(settings_cls, json_file=[p5, p6], deep_merge=deep_merge),)
+
+    s = Settings()
+    assert s.model_dump() == {'hello': 'world', 'nested': {'foo': 3, 'bar': 2 if deep_merge else 0}}

tests/test_source_toml.py

@@ -114,3 +114,47 @@ def settings_customise_sources(
 
     s = Settings()
     assert s.model_dump() == {'toml1': 1, 'toml2': 2}
+
+
+@pytest.mark.skipif(sys.version_info <= (3, 11) and tomli is None, reason='tomli/tomllib is not installed')
+@pytest.mark.parametrize('deep_merge', [False, True])
+def test_multiple_file_toml_merge(tmp_path, deep_merge):
+    p1 = tmp_path / '.env.toml1'
+    p2 = tmp_path / '.env.toml2'
+    p1.write_text(
+        """
+    hello = "world"
+
+    [nested]
+    foo=1
+    bar=2
+    """
+    )
+    p2.write_text(
+        """
+    [nested]
+    foo=3
+    """
+    )
+
+    class Nested(BaseModel):
+        foo: int
+        bar: int = 0
+
+    class Settings(BaseSettings):
+        hello: str
+        nested: Nested
+
+        @classmethod
+        def settings_customise_sources(
+            cls,
+            settings_cls: type[BaseSettings],
+            init_settings: PydanticBaseSettingsSource,
+            env_settings: PydanticBaseSettingsSource,
+            dotenv_settings: PydanticBaseSettingsSource,
+            file_secret_settings: PydanticBaseSettingsSource,
+        ) -> tuple[PydanticBaseSettingsSource, ...]:
+            return (TomlConfigSettingsSource(settings_cls, toml_file=[p1, p2], deep_merge=deep_merge),)
+
+    s = Settings()
+    assert s.model_dump() == {'hello': 'world', 'nested': {'foo': 3, 'bar': 2 if deep_merge else 0}}

tests/test_source_yaml.py

@@ -167,6 +167,50 @@ def settings_customise_sources(
     assert s.model_dump() == {'yaml3': 3, 'yaml4': 4}
 
 
+@pytest.mark.skipif(yaml is None, reason='pyYAML is not installed')
+@pytest.mark.parametrize('deep_merge', [False, True])
+def test_multiple_file_yaml_deep_merge(tmp_path, deep_merge):
+    p3 = tmp_path / '.env.yaml3'
+    p4 = tmp_path / '.env.yaml4'
+    p3.write_text(
+        """
+    hello: world
+
+    nested:
+      foo: 1
+      bar: 2
+    """
+    )
+    p4.write_text(
+        """
+    nested:
+      foo: 3
+    """
+    )
+
+    class Nested(BaseModel):
+        foo: int
+        bar: int = 0
+
+    class Settings(BaseSettings):
+        hello: str
+        nested: Nested
+
+        @classmethod
+        def settings_customise_sources(
+            cls,
+            settings_cls: type[BaseSettings],
+            init_settings: PydanticBaseSettingsSource,
+            env_settings: PydanticBaseSettingsSource,
+            dotenv_settings: PydanticBaseSettingsSource,
+            file_secret_settings: PydanticBaseSettingsSource,
+        ) -> tuple[PydanticBaseSettingsSource, ...]:
+            return (YamlConfigSettingsSource(settings_cls, yaml_file=[p3, p4], deep_merge=deep_merge),)
+
+    s = Settings()
+    assert s.model_dump() == {'hello': 'world', 'nested': {'foo': 3, 'bar': 2 if deep_merge else 0}}
+
+
 @pytest.mark.skipif(yaml is None, reason='pyYAML is not installed')
 def test_yaml_config_section(tmp_path):
     p = tmp_path / '.env'