Apply source order: init > env > dotenv > secrets > defaults and preserve earlier values

Johannes Rueschel · Johannes Rueschel · commit b534e7cb6ce2 · 2025-09-30T15:17:59.000+02:00
- Deep merge for config files using deep_update in ConfigFileSourceMixin._read_files
- Deterministic alias selection for init kwargs (prefer first alias in validation_alias; field name allowed with populate_by_name=True).
diff --git a/pydantic_settings/sources/base.py b/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_extensions import get_args
 from typing_inspection import typing_objects
@@ -202,7 +202,9 @@ def _read_files(self, files: PathType | None) -> dict[str, Any]:
         for file in files:
             file_path = Path(file).expanduser()
             if file_path.is_file():
-                vars.update(self._read_file(file_path))
+                file_data = self._read_file(file_path)
+                # Deep merge so later files override earlier nested keys instead of replacing whole objects
+                vars = deep_update(vars, file_data)
         return vars
 
     @abstractmethod
@@ -265,12 +267,27 @@ def __init__(
         init_kwarg_names = set(init_kwargs.keys())
         for field_name, field_info in settings_cls.model_fields.items():
             alias_names, *_ = _get_alias_names(field_name, field_info)
-            init_kwarg_name = init_kwarg_names & set(alias_names)
+            # When populate_by_name is True, allow using the field name as an input key,
+            # but normalize to the preferred alias to keep keys consistent across sources.
+            matchable_names = set(alias_names)
+            include_name = settings_cls.model_config.get('populate_by_name', False)
+            if include_name:
+                matchable_names.add(field_name)
+            init_kwarg_name = init_kwarg_names & matchable_names
             if init_kwarg_name:
-                preferred_alias = alias_names[0]
-                preferred_set_alias = next(alias for alias in alias_names if alias in init_kwarg_name)
+                preferred_alias = alias_names[0] if alias_names else field_name
+                # Choose provided key deterministically: prefer the first alias in alias_names order;
+                # fall back to field_name if allowed and provided.
+                provided_key = next((alias for alias in alias_names if alias in init_kwarg_names), None)
+                if provided_key is None and include_name and field_name in init_kwarg_names:
+                    provided_key = field_name
+                # provided_key should not be None here because init_kwarg_name is non-empty
+                assert provided_key is not None
                 init_kwarg_names -= init_kwarg_name
-                self.init_kwargs[preferred_alias] = init_kwargs[preferred_set_alias]
+                self.init_kwargs[preferred_alias] = init_kwargs[provided_key]
+        # Include any remaining init kwargs (e.g., extras) unchanged
+        # Note: If populate_by_name is True and the provided key is the field name, but
+        # no alias exists, we keep it as-is so it can be processed as extra if allowed.
         self.init_kwargs.update({key: val for key, val in init_kwargs.items() if key in init_kwarg_names})
 
         super().__init__(settings_cls)
diff --git a/tests/test_config_file_deep_merge.py b/tests/test_config_file_deep_merge.py
@@ -0,0 +1,203 @@
+from __future__ import annotations as _annotations
+
+import json
+import sys
+from pathlib import Path
+from typing import Optional
+
+import pytest
+from pydantic import AnyHttpUrl, Field
+
+try:
+    import yaml  # type: ignore
+except Exception:
+    yaml = None
+
+try:
+    import tomli  # type: ignore
+except Exception:
+    tomli = None
+
+from pydantic_settings import (
+    BaseSettings,
+    JsonConfigSettingsSource,
+    PydanticBaseSettingsSource,
+    SettingsConfigDict,
+    TomlConfigSettingsSource,
+    YamlConfigSettingsSource,
+)
+
+
+def test_init_kwargs_override_env_for_alias_with_populate_by_name(monkeypatch):
+    class Settings(BaseSettings):
+        abc: AnyHttpUrl = Field(validation_alias='my_abc')
+        model_config = SettingsConfigDict(populate_by_name=True, extra='allow')
+
+    monkeypatch.setenv('MY_ABC', 'http://localhost.com/')
+
+    # Passing by field name should be accepted (populate_by_name=True) and should
+    # override env-derived value. Also ensures init > env precedence with validation_alias.
+    assert str(Settings(abc='http://prod.localhost.com/').abc) == 'http://prod.localhost.com/'
+
+
+def test_deep_merge_multiple_file_json(tmp_path: Path):
+    p1 = tmp_path / 'a.json'
+    p2 = tmp_path / 'b.json'
+
+    with open(p1, 'w') as f1:
+        json.dump({'a': 1, 'nested': {'x': 1, 'y': 1}}, f1)
+    with open(p2, 'w') as f2:
+        json.dump({'b': 2, 'nested': {'y': 2, 'z': 3}}, f2)
+
+    class Settings(BaseSettings):
+        a: Optional[int] = None
+        b: Optional[int] = None
+        nested: dict[str, int]
+
+        @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=[p1, p2]),)
+
+    s = Settings()
+    assert s.a == 1
+    assert s.b == 2
+    assert s.nested == {'x': 1, 'y': 2, 'z': 3}
+
+
+@pytest.mark.skipif(yaml is None, reason='pyYAML is not installed')
+def test_deep_merge_multiple_file_yaml(tmp_path: Path):
+    p1 = tmp_path / 'a.yaml'
+    p2 = tmp_path / 'b.yaml'
+
+    p1.write_text(
+        """
+    a: 1
+    nested:
+      x: 1
+      y: 1
+    """
+    )
+    p2.write_text(
+        """
+    b: 2
+    nested:
+      y: 2
+      z: 3
+    """
+    )
+
+    class Settings(BaseSettings):
+        a: Optional[int] = None
+        b: Optional[int] = None
+        nested: dict[str, int]
+
+        @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=[p1, p2]),)
+
+    s = Settings()
+    assert s.a == 1
+    assert s.b == 2
+    assert s.nested == {'x': 1, 'y': 2, 'z': 3}
+
+
+@pytest.mark.skipif(sys.version_info <= (3, 11) and tomli is None, reason='tomli/tomllib is not installed')
+def test_deep_merge_multiple_file_toml(tmp_path: Path):
+    p1 = tmp_path / 'a.toml'
+    p2 = tmp_path / 'b.toml'
+
+    p1.write_text(
+        """
+    a=1
+
+    [nested]
+    x=1
+    y=1
+    """
+    )
+    p2.write_text(
+        """
+    b=2
+
+    [nested]
+    y=2
+    z=3
+    """
+    )
+
+    class Settings(BaseSettings):
+        a: Optional[int] = None
+        b: Optional[int] = None
+        nested: dict[str, int]
+
+        @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]),)
+
+    s = Settings()
+    assert s.a == 1
+    assert s.b == 2
+    assert s.nested == {'x': 1, 'y': 2, 'z': 3}
+
+
+@pytest.mark.skipif(yaml is None, reason='pyYAML is not installed')
+def test_yaml_config_section_after_deep_merge(tmp_path: Path):
+    # Ensure that config section is picked from the deep-merged data
+    p1 = tmp_path / 'a.yaml'
+    p2 = tmp_path / 'b.yaml'
+    p1.write_text(
+        """
+    nested:
+      x: 1
+      y: 1
+    """
+    )
+    p2.write_text(
+        """
+    nested:
+      y: 2
+      z: 3
+    other: true
+    """
+    )
+
+    class S2(BaseSettings):
+        x: int
+        y: int
+        z: int
+        model_config = SettingsConfigDict(yaml_config_section='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=[p1, p2]),)
+
+    s2 = S2()
+    assert s2.model_dump() == {'x': 1, 'y': 2, 'z': 3}
diff --git a/tests/test_precedence_and_merging.py b/tests/test_precedence_and_merging.py
@@ -0,0 +1,132 @@
+from __future__ import annotations as _annotations
+
+from pathlib import Path
+
+import pytest
+from pydantic import AnyHttpUrl, Field
+
+from pydantic_settings import (
+    BaseSettings,
+    PydanticBaseSettingsSource,
+    SettingsConfigDict,
+)
+
+
+@pytest.fixture(autouse=True)
+def clear_env(monkeypatch):
+    monkeypatch.delenv('FOO', raising=False)
+    monkeypatch.delenv('BAR', raising=False)
+    monkeypatch.delenv('NESTED', raising=False)
+    monkeypatch.delenv('NESTED__X', raising=False)
+    monkeypatch.delenv('NESTED__Y', raising=False)
+
+
+def test_init_kwargs_override_env_for_alias_with_populate_by_name(monkeypatch):
+    class Settings(BaseSettings):
+        abc: AnyHttpUrl = Field(validation_alias='my_abc')
+        model_config = SettingsConfigDict(populate_by_name=True, extra='allow')
+
+    monkeypatch.setenv('MY_ABC', 'http://localhost.com/')
+
+    # Passing by field name should be accepted (populate_by_name=True) and should
+    # override env-derived value. Also ensures init > env precedence with validation_alias.
+    assert str(Settings(abc='http://prod.localhost.com/').abc) == 'http://prod.localhost.com/'
+
+
+def test_precedence_init_over_env(tmp_path: Path, monkeypatch):
+    class Settings(BaseSettings):
+        foo: str
+
+    monkeypatch.setenv('FOO', 'from-env')
+
+    # init should win over env
+    s = Settings(foo='from-init')
+    assert s.foo == 'from-init'
+
+
+def test_precedence_env_over_dotenv(tmp_path: Path, monkeypatch):
+    env_file = tmp_path / '.env'
+    env_file.write_text('FOO=from-dotenv\n')
+
+    class Settings(BaseSettings):
+        foo: str
+
+        model_config = SettingsConfigDict(env_file=env_file)
+
+    # env set should override dotenv
+    monkeypatch.setenv('FOO', 'from-env')
+    s = Settings()
+    assert s.foo == 'from-env'
+
+
+def test_precedence_dotenv_over_secrets(tmp_path: Path, monkeypatch):
+    # create dotenv
+    env_file = tmp_path / '.env'
+    env_file.write_text('FOO=from-dotenv\n')
+
+    # create secrets directory with same key
+    secrets_dir = tmp_path / 'secrets'
+    secrets_dir.mkdir()
+    (secrets_dir / 'FOO').write_text('from-secrets\n')
+
+    class Settings(BaseSettings):
+        foo: str
+
+        model_config = SettingsConfigDict(env_file=env_file, secrets_dir=secrets_dir)
+
+    # No env set, dotenv should override secrets
+    s = Settings()
+    assert s.foo == 'from-dotenv'
+
+
+def test_precedence_secrets_over_defaults(tmp_path: Path):
+    secrets_dir = tmp_path / 'secrets'
+    secrets_dir.mkdir()
+    (secrets_dir / 'FOO').write_text('from-secrets\n')
+
+    class Settings(BaseSettings):
+        foo: str = 'from-default'
+
+        model_config = SettingsConfigDict(secrets_dir=secrets_dir)
+
+    s = Settings()
+    assert s.foo == 'from-secrets'
+
+
+def test_merging_preserves_earlier_values(tmp_path: Path, monkeypatch):
+    # Prove that merging preserves earlier source values: init -> env -> dotenv -> secrets -> defaults
+    # We'll populate nested from dotenv and env parts, then set a default for a, and init for b
+    env_file = tmp_path / '.env'
+    env_file.write_text('NESTED={"x":1}\n')
+
+    secrets_dir = tmp_path / 'secrets'
+    secrets_dir.mkdir()
+    (secrets_dir / 'NESTED').write_text('{"y": 2}')
+
+    class Settings(BaseSettings):
+        a: int = 10
+        b: int = 0
+        nested: dict
+
+        model_config = SettingsConfigDict(env_file=env_file, secrets_dir=secrets_dir, env_nested_delimiter='__')
+
+        @classmethod
+        def settings_customise_sources(
+            cls,
+            settings_cls: type[BaseSettings],
+            init_settings: PydanticBaseSettingsSource,
+            env_settings: PydanticBaseSettingsSource,
+            dotenv_settings: PydanticBaseSettingsSource,
+            file_secret_settings: PydanticBaseSettingsSource,
+        ):
+            # normal order; we want to assert deep merging
+            return init_settings, env_settings, dotenv_settings, file_secret_settings
+
+    # env contributes nested.y and overrides dotenv nested.x=1 if set; we'll set only y to prove merge
+    monkeypatch.setenv('NESTED__y', '3')
+    # init contributes b, defaults contribute a
+    s = Settings(b=20)
+    assert s.a == 10  # defaults preserved
+    assert s.b == 20  # init wins
+    # nested: dotenv provides x=1; env provides y=3; deep merged => {x:1, y:3}
+    assert s.nested == {'x': 1, 'y': 3}
