Initial Commit

2026-10-21 13:04:35 +01:00
commit 2c86cd435c
4963 changed files with 779038 additions and 0 deletions
@@ -0,0 +1,146 @@
+"""Secrets file settings source."""
+
+from __future__ import annotations as _annotations
+
+import logging
+import os
+import warnings
+from pathlib import Path
+from typing import (
+    TYPE_CHECKING,
+    Any,
+)
+
+from pydantic.fields import FieldInfo
+
+from pydantic_settings.utils import _settings_debug_enabled, logger, path_type_label
+
+from ...exceptions import SettingsError
+from ..base import PydanticBaseEnvSettingsSource
+from ..types import EnvPrefixTarget, PathType
+from ..utils import InitState
+
+if TYPE_CHECKING:
+    from pydantic_settings.main import BaseSettings
+
+
+class SecretsSettingsSource(PydanticBaseEnvSettingsSource):
+    """
+    Source class for loading settings values from secret files.
+    """
+
+    def __init__(
+        self,
+        settings_cls: type[BaseSettings],
+        secrets_dir: PathType | None = None,
+        case_sensitive: bool | None = None,
+        env_prefix: str | None = None,
+        env_prefix_target: EnvPrefixTarget | None = None,
+        env_ignore_empty: bool | None = None,
+        env_parse_none_str: str | None = None,
+        env_parse_enums: bool | None = None,
+        _init_state: InitState | None = None,
+    ) -> None:
+        super().__init__(
+            settings_cls,
+            case_sensitive,
+            env_prefix,
+            env_prefix_target,
+            env_ignore_empty,
+            env_parse_none_str,
+            env_parse_enums,
+            _init_state,
+        )
+        self.secrets_dir = secrets_dir if secrets_dir is not None else self.config.get('secrets_dir')
+
+    def __call__(self) -> dict[str, Any]:
+        """
+        Build fields from "secrets" files.
+        """
+        secrets: dict[str, str | None] = {}
+
+        if self.secrets_dir is None:
+            return secrets
+
+        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 = []
+
+        for path in secrets_paths:
+            if not path.exists():
+                # No `stacklevel`: this source is reached at different depths — via
+                # `Settings()` (through `_settings_build_values`) or by instantiating
+                # the source directly — so no fixed value attributes the warning to
+                # user code in both cases.
+                warnings.warn(f'directory "{path}" does not exist')  # noqa: B028
+            else:
+                self.secrets_paths.append(path)
+
+        if not len(self.secrets_paths):
+            return secrets
+
+        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__()
+
+    @classmethod
+    def find_case_path(cls, dir_path: Path, file_name: str, case_sensitive: bool) -> Path | None:
+        """
+        Find a file within path's directory matching filename, optionally ignoring case.
+
+        Args:
+            dir_path: Directory path.
+            file_name: File name.
+            case_sensitive: Whether to search for file name case sensitively.
+
+        Returns:
+            Whether file path or `None` if file does not exist in directory.
+        """
+        for f in dir_path.iterdir():
+            if f.name == file_name or (not case_sensitive and f.name.lower() == file_name.lower()):
+                return f
+        return None
+
+    def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
+        """
+        Gets the value for field from secret file and a flag to determine whether value is complex.
+
+        Args:
+            field: The field.
+            field_name: The field name.
+
+        Returns:
+            A tuple that contains the value (`None` if the file does not exist), key, and
+                a flag to determine whether value is complex.
+        """
+
+        debug = _settings_debug_enabled() and logger.isEnabledFor(logging.DEBUG)
+
+        for field_key, env_name, value_is_complex in self._extract_field_info(field, field_name):
+            # 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
+                    if debug:
+                        logger.debug('Secret file not found, skipping: %s', (secrets_path / env_name).resolve())
+                    continue
+
+                if path.is_file():
+                    if debug:
+                        logger.debug('Loading secret file: %s', path.resolve())
+                    # Explicit encoding: `read_text()` would otherwise decode with the
+                    # locale encoding, so a UTF-8 secret is corrupted on a non-UTF-8
+                    # Windows code page (silently, when every byte happens to map).
+                    return path.read_text(encoding='utf-8').strip(), field_key, value_is_complex
+                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
+
+    def __repr__(self) -> str:
+        return f'{self.__class__.__name__}(secrets_dir={self.secrets_dir!r})'