From 12d85cf521bf2b07939241bf33f17d14d906623f Mon Sep 17 00:00:00 2001 From: Michael Makukha Date: Sat, 7 Sep 2024 20:26:34 +0300 Subject: [PATCH] Enable multiple secrets dirs (#372) Co-authored-by: Hasan Ramezani --- .gitignore | 2 + docs/index.md | 16 +++++ pydantic_settings/main.py | 9 ++- pydantic_settings/sources.py | 44 ++++++++------ tests/test_settings.py | 113 +++++++++++++++++++++++++++++++++++ 5 files changed, 162 insertions(+), 22 deletions(-) diff --git a/.gitignore b/.gitignore index 2079909a..251e2c11 100644 --- a/.gitignore +++ b/.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 diff --git a/docs/index.md b/docs/index.md index 5e3cf281..4d0c9fe0 100644 --- a/docs/index.md +++ b/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. diff --git a/pydantic_settings/main.py b/pydantic_settings/main.py index 33ea245c..19d97eaa 100644 --- a/pydantic_settings/main.py +++ b/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') diff --git a/pydantic_settings/sources.py b/pydantic_settings/sources.py index c5cb5c44..38cc79d8 100644 --- a/pydantic_settings/sources.py +++ b/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 diff --git a/tests/test_settings.py b/tests/test_settings.py index bff09b40..c5813955 100644 --- a/tests/test_settings.py +++ b/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')