Enable multiple secrets dirs (#372)

Co-authored-by: Hasan Ramezani <[email protected]>

Author: makukha

.gitignore

@@ -1,5 +1,6 @@
 .idea/
 env/
+.envrc
 venv/
 .venv/
 env3*/
@@ -18,6 +19,7 @@ test.py
 /site/
 /site.zip
 .pytest_cache/
+.python-version
 .vscode/
 _build/
 .auto-format

docs/index.md

@@ -1214,6 +1214,22 @@ Even when using a secrets directory, *pydantic* will still read environment vari
 Passing a file path via the `_secrets_dir` keyword argument on instantiation (method 2) will override
 the value (if any) set on the `model_config` class.
 
+If you need to load settings from multiple secrets directories, you can pass multiple paths as a tuple or list. Just like for `env_file`, values from subsequent paths override previous ones.
+
+````python
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    # files in '/run/secrets' take priority over '/var/run'
+    model_config = SettingsConfigDict(secrets_dir=('/var/run', '/run/secrets'))
+
+    database_password: str
+````
+
+If any of `secrets_dir` is missing, it is ignored, and warning is shown. If any of `secrets_dir` is a file, error is raised.
+
+
 ### Use Case: Docker Secrets
 
 Docker Secrets can be used to provide sensitive configuration to an application running in a Docker container.

pydantic_settings/main.py

@@ -1,6 +1,5 @@
 from __future__ import annotations as _annotations
 
-from pathlib import Path
 from typing import Any, ClassVar
 
 from pydantic import ConfigDict
@@ -43,7 +42,7 @@ class SettingsConfigDict(ConfigDict, total=False):
     cli_exit_on_error: bool
     cli_prefix: str
     cli_implicit_flags: bool | None
-    secrets_dir: str | Path | None
+    secrets_dir: PathType | None
     json_file: PathType | None
     json_file_encoding: str | None
     yaml_file: PathType | None
@@ -121,7 +120,7 @@ class BaseSettings(BaseModel):
         _cli_prefix: The root parser command line arguments prefix. Defaults to "".
         _cli_implicit_flags: Whether `bool` fields should be implicitly converted into CLI boolean flags.
             (e.g. --flag, --no-flag). Defaults to `False`.
-        _secrets_dir: The secret files directory. Defaults to `None`.
+        _secrets_dir: The secret files directory or a sequence of directories. Defaults to `None`.
     """
 
     def __init__(
@@ -146,7 +145,7 @@ def __init__(
         _cli_exit_on_error: bool | None = None,
         _cli_prefix: str | None = None,
         _cli_implicit_flags: bool | None = None,
-        _secrets_dir: str | Path | None = None,
+        _secrets_dir: PathType | None = None,
         **values: Any,
     ) -> None:
         # Uses something other than `self` the first arg to allow "self" as a settable attribute
@@ -224,7 +223,7 @@ def _settings_build_values(
         _cli_exit_on_error: bool | None = None,
         _cli_prefix: str | None = None,
         _cli_implicit_flags: bool | None = None,
-        _secrets_dir: str | Path | None = None,
+        _secrets_dir: PathType | None = None,
     ) -> dict[str, Any]:
         # Determine settings config values
         case_sensitive = _case_sensitive if _case_sensitive is not None else self.model_config.get('case_sensitive')

pydantic_settings/sources.py

@@ -574,7 +574,7 @@ class SecretsSettingsSource(PydanticBaseEnvSettingsSource):
     def __init__(
         self,
         settings_cls: type[BaseSettings],
-        secrets_dir: str | Path | None = None,
+        secrets_dir: PathType | None = None,
         case_sensitive: bool | None = None,
         env_prefix: str | None = None,
         env_ignore_empty: bool | None = None,
@@ -595,14 +595,22 @@ def __call__(self) -> dict[str, Any]:
         if self.secrets_dir is None:
             return secrets
 
-        self.secrets_path = Path(self.secrets_dir).expanduser()
+        secrets_dirs = [self.secrets_dir] if isinstance(self.secrets_dir, (str, os.PathLike)) else self.secrets_dir
+        secrets_paths = [Path(p).expanduser() for p in secrets_dirs]
+        self.secrets_paths = []
 
-        if not self.secrets_path.exists():
-            warnings.warn(f'directory "{self.secrets_path}" does not exist')
+        for path in secrets_paths:
+            if not path.exists():
+                warnings.warn(f'directory "{path}" does not exist')
+            else:
+                self.secrets_paths.append(path)
+
+        if not len(self.secrets_paths):
             return secrets
 
-        if not self.secrets_path.is_dir():
-            raise SettingsError(f'secrets_dir must reference a directory, not a {path_type_label(self.secrets_path)}')
+        for path in self.secrets_paths:
+            if not path.is_dir():
+                raise SettingsError(f'secrets_dir must reference a directory, not a {path_type_label(path)}')
 
         return super().__call__()
 
@@ -640,18 +648,20 @@ def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str,
         """
 
         for field_key, env_name, value_is_complex in self._extract_field_info(field, field_name):
-            path = self.find_case_path(self.secrets_path, env_name, self.case_sensitive)
-            if not path:
-                # path does not exist, we currently don't return a warning for this
-                continue
+            # paths reversed to match the last-wins behaviour of `env_file`
+            for secrets_path in reversed(self.secrets_paths):
+                path = self.find_case_path(secrets_path, env_name, self.case_sensitive)
+                if not path:
+                    # path does not exist, we currently don't return a warning for this
+                    continue
 
-            if path.is_file():
-                return path.read_text().strip(), field_key, value_is_complex
-            else:
-                warnings.warn(
-                    f'attempted to load secret file "{path}" but found a {path_type_label(path)} instead.',
-                    stacklevel=4,
-                )
+                if path.is_file():
+                    return path.read_text().strip(), field_key, value_is_complex
+                else:
+                    warnings.warn(
+                        f'attempted to load secret file "{path}" but found a {path_type_label(path)} instead.',
+                        stacklevel=4,
+                    )
 
         return None, field_key, value_is_complex
 

tests/test_settings.py

@@ -1416,6 +1416,33 @@ class Settings(BaseSettings):
     assert Settings().model_dump() == {'foo': 'foo_secret_value_str'}
 
 
+def test_secrets_path_multiple(tmp_path):
+    d1 = tmp_path / 'dir1'
+    d2 = tmp_path / 'dir2'
+    d1.mkdir()
+    d2.mkdir()
+    (d1 / 'foo1').write_text('foo1_dir1_secret_value_str')
+    (d1 / 'foo2').write_text('foo2_dir1_secret_value_str')
+    (d2 / 'foo2').write_text('foo2_dir2_secret_value_str')
+    (d2 / 'foo3').write_text('foo3_dir2_secret_value_str')
+
+    class Settings(BaseSettings):
+        foo1: str
+        foo2: str
+        foo3: str
+
+    assert Settings(_secrets_dir=(d1, d2)).model_dump() == {
+        'foo1': 'foo1_dir1_secret_value_str',
+        'foo2': 'foo2_dir2_secret_value_str',  # dir2 takes priority
+        'foo3': 'foo3_dir2_secret_value_str',
+    }
+    assert Settings(_secrets_dir=(d2, d1)).model_dump() == {
+        'foo1': 'foo1_dir1_secret_value_str',
+        'foo2': 'foo2_dir1_secret_value_str',  # dir1 takes priority
+        'foo3': 'foo3_dir2_secret_value_str',
+    }
+
+
 def test_secrets_path_with_validation_alias(tmp_path):
     p = tmp_path / 'foo'
     p.write_text('{"bar": ["test"]}')
@@ -1535,6 +1562,28 @@ class Settings(BaseSettings):
         Settings()
 
 
+def test_secrets_invalid_secrets_dir_multiple_all(tmp_path):
+    class Settings(BaseSettings):
+        foo: str
+
+    (d1 := tmp_path / 'dir1').write_text('')
+    (d2 := tmp_path / 'dir2').write_text('')
+
+    with pytest.raises(SettingsError, match='secrets_dir must reference a directory, not a file'):
+        Settings(_secrets_dir=[d1, d2])
+
+
+def test_secrets_invalid_secrets_dir_multiple_one(tmp_path):
+    class Settings(BaseSettings):
+        foo: str
+
+    (d1 := tmp_path / 'dir1').mkdir()
+    (d2 := tmp_path / 'dir2').write_text('')
+
+    with pytest.raises(SettingsError, match='secrets_dir must reference a directory, not a file'):
+        Settings(_secrets_dir=[d1, d2])
+
+
 @pytest.mark.skipif(sys.platform.startswith('win'), reason='windows paths break regex')
 def test_secrets_missing_location(tmp_path):
     class Settings(BaseSettings):
@@ -1544,6 +1593,34 @@ class Settings(BaseSettings):
         Settings()
 
 
+@pytest.mark.skipif(sys.platform.startswith('win'), reason='windows paths break regex')
+def test_secrets_missing_location_multiple_all(tmp_path):
+    class Settings(BaseSettings):
+        foo: Optional[str] = None
+
+    with pytest.warns() as record:
+        Settings(_secrets_dir=[tmp_path / 'dir1', tmp_path / 'dir2'])
+
+    assert len(record) == 2
+    assert record[0].category is UserWarning and record[1].category is UserWarning
+    assert str(record[0].message) == f'directory "{tmp_path}/dir1" does not exist'
+    assert str(record[1].message) == f'directory "{tmp_path}/dir2" does not exist'
+
+
+@pytest.mark.skipif(sys.platform.startswith('win'), reason='windows paths break regex')
+def test_secrets_missing_location_multiple_one(tmp_path):
+    class Settings(BaseSettings):
+        foo: Optional[str] = None
+
+    (d1 := tmp_path / 'dir1').mkdir()
+    (d1 / 'foo').write_text('secret_value')
+
+    with pytest.warns(UserWarning, match=f'directory "{tmp_path}/dir2" does not exist'):
+        conf = Settings(_secrets_dir=[d1, tmp_path / 'dir2'])
+
+    assert conf.foo == 'secret_value'  # value obtained from first directory
+
+
 @pytest.mark.skipif(sys.platform.startswith('win'), reason='windows paths break regex')
 def test_secrets_file_is_a_directory(tmp_path):
     p1 = tmp_path / 'foo'
@@ -1558,6 +1635,42 @@ class Settings(BaseSettings):
         Settings()
 
 
+@pytest.mark.skipif(sys.platform.startswith('win'), reason='windows paths break regex')
+def test_secrets_file_is_a_directory_multiple_all(tmp_path):
+    class Settings(BaseSettings):
+        foo: Optional[str] = None
+
+    (d1 := tmp_path / 'dir1').mkdir()
+    (d2 := tmp_path / 'dir2').mkdir()
+    (d1 / 'foo').mkdir()
+    (d2 / 'foo').mkdir()
+
+    with pytest.warns() as record:
+        Settings(_secrets_dir=[d1, d2])
+
+    assert len(record) == 2
+    assert record[0].category is UserWarning and record[1].category is UserWarning
+    # warnings are emitted in reverse order
+    assert str(record[0].message) == f'attempted to load secret file "{d2}/foo" but found a directory instead.'
+    assert str(record[1].message) == f'attempted to load secret file "{d1}/foo" but found a directory instead.'
+
+
+@pytest.mark.skipif(sys.platform.startswith('win'), reason='windows paths break regex')
+def test_secrets_file_is_a_directory_multiple_one(tmp_path):
+    class Settings(BaseSettings):
+        foo: Optional[str] = None
+
+    (d1 := tmp_path / 'dir1').mkdir()
+    (d2 := tmp_path / 'dir2').mkdir()
+    (d1 / 'foo').write_text('secret_value')
+    (d2 / 'foo').mkdir()
+
+    with pytest.warns(UserWarning, match=f'attempted to load secret file "{d2}/foo" but found a directory instead.'):
+        conf = Settings(_secrets_dir=[d1, d2])
+
+    assert conf.foo == 'secret_value'  # value obtained from first directory
+
+
 def test_secrets_dotenv_precedence(tmp_path):
     s = tmp_path / 'foo'
     s.write_text('foo_secret_value_str')