"""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(): warnings.warn(f'directory "{path}" does not exist') 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 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 def __repr__(self) -> str: return f'{self.__class__.__name__}(secrets_dir={self.secrets_dir!r})'