From fe02a2224039b3b62fad3cfc2b6310c7f9a0698c Mon Sep 17 00:00:00 2001 From: Polina Date: Thu, 2 Jul 2026 20:11:30 +0000 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=B8?= =?UTF-8?q?=D1=82=D1=8C=20=D1=84=D0=B0=D0=B9=D0=BB=D1=8B=20=D0=B2=20=C2=AB?= =?UTF-8?q?venv/Lib/site-packages/pydantic=5Fsettings/sources/providers?= =?UTF-8?q?=C2=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../sources/providers/secrets.py | 132 ++++++++++++++++++ .../sources/providers/toml.py | 67 +++++++++ .../sources/providers/yaml.py | 130 +++++++++++++++++ 3 files changed, 329 insertions(+) create mode 100644 venv/Lib/site-packages/pydantic_settings/sources/providers/secrets.py create mode 100644 venv/Lib/site-packages/pydantic_settings/sources/providers/toml.py create mode 100644 venv/Lib/site-packages/pydantic_settings/sources/providers/yaml.py diff --git a/venv/Lib/site-packages/pydantic_settings/sources/providers/secrets.py b/venv/Lib/site-packages/pydantic_settings/sources/providers/secrets.py new file mode 100644 index 0000000..14364c7 --- /dev/null +++ b/venv/Lib/site-packages/pydantic_settings/sources/providers/secrets.py @@ -0,0 +1,132 @@ +"""Secrets file settings source.""" + +from __future__ import annotations as _annotations + +import os +import warnings +from pathlib import Path +from typing import ( + TYPE_CHECKING, + Any, +) + +from pydantic.fields import FieldInfo + +from pydantic_settings.utils import path_type_label + +from ...exceptions import SettingsError +from ..base import PydanticBaseEnvSettingsSource +from ..types import EnvPrefixTarget, PathType + +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, + ) -> None: + super().__init__( + settings_cls, + case_sensitive, + env_prefix, + env_prefix_target, + env_ignore_empty, + env_parse_none_str, + env_parse_enums, + ) + 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: + return f + elif 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. + """ + + 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 + 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, + ) + + return None, field_key, value_is_complex + + def __repr__(self) -> str: + return f'{self.__class__.__name__}(secrets_dir={self.secrets_dir!r})' diff --git a/venv/Lib/site-packages/pydantic_settings/sources/providers/toml.py b/venv/Lib/site-packages/pydantic_settings/sources/providers/toml.py new file mode 100644 index 0000000..8de9f28 --- /dev/null +++ b/venv/Lib/site-packages/pydantic_settings/sources/providers/toml.py @@ -0,0 +1,67 @@ +"""TOML file settings source.""" + +from __future__ import annotations as _annotations + +import sys +from pathlib import Path +from typing import ( + TYPE_CHECKING, + Any, +) + +from ..base import ConfigFileSourceMixin, InitSettingsSource +from ..types import DEFAULT_PATH, PathType + +if TYPE_CHECKING: + from pydantic_settings.main import BaseSettings + + if sys.version_info >= (3, 11): + import tomllib + else: + tomllib = None + import tomli +else: + tomllib = None + tomli = None + + +def import_toml() -> None: + global tomli + global tomllib + if sys.version_info < (3, 11): + if tomli is not None: + return + try: + import tomli + except ImportError as e: # pragma: no cover + raise ImportError('tomli is not installed, run `pip install pydantic-settings[toml]`') from e + else: + if tomllib is not None: + return + import tomllib + + +class TomlConfigSettingsSource(InitSettingsSource, ConfigFileSourceMixin): + """ + A source class that loads variables from a TOML file + """ + + def __init__( + self, + settings_cls: type[BaseSettings], + toml_file: PathType | None = DEFAULT_PATH, + deep_merge: bool = False, + ): + self.toml_file_path = toml_file if toml_file != DEFAULT_PATH else settings_cls.model_config.get('toml_file') + self.toml_data = self._read_files(self.toml_file_path, deep_merge=deep_merge) + super().__init__(settings_cls, self.toml_data) + + def _read_file(self, file_path: Path) -> dict[str, Any]: + import_toml() + with file_path.open(mode='rb') as toml_file: + if sys.version_info < (3, 11): + return tomli.load(toml_file) + return tomllib.load(toml_file) + + def __repr__(self) -> str: + return f'{self.__class__.__name__}(toml_file={self.toml_file_path})' diff --git a/venv/Lib/site-packages/pydantic_settings/sources/providers/yaml.py b/venv/Lib/site-packages/pydantic_settings/sources/providers/yaml.py new file mode 100644 index 0000000..e3e48d0 --- /dev/null +++ b/venv/Lib/site-packages/pydantic_settings/sources/providers/yaml.py @@ -0,0 +1,130 @@ +"""YAML file settings source.""" + +from __future__ import annotations as _annotations + +from pathlib import Path +from typing import ( + TYPE_CHECKING, + Any, +) + +from ..base import ConfigFileSourceMixin, InitSettingsSource +from ..types import DEFAULT_PATH, PathType + +if TYPE_CHECKING: + import yaml + + from pydantic_settings.main import BaseSettings +else: + yaml = None + + +def import_yaml() -> None: + global yaml + if yaml is not None: + return + try: + import yaml + except ImportError as e: + raise ImportError('PyYAML is not installed, run `pip install pydantic-settings[yaml]`') from e + + +class YamlConfigSettingsSource(InitSettingsSource, ConfigFileSourceMixin): + """ + A source class that loads variables from a yaml file + """ + + def __init__( + self, + settings_cls: type[BaseSettings], + yaml_file: PathType | None = DEFAULT_PATH, + yaml_file_encoding: str | None = None, + yaml_config_section: str | None = None, + deep_merge: bool = False, + ): + self.yaml_file_path = yaml_file if yaml_file != DEFAULT_PATH else settings_cls.model_config.get('yaml_file') + self.yaml_file_encoding = ( + yaml_file_encoding + if yaml_file_encoding is not None + else settings_cls.model_config.get('yaml_file_encoding') + ) + self.yaml_config_section = ( + yaml_config_section + if yaml_config_section is not None + else settings_cls.model_config.get('yaml_config_section') + ) + self.yaml_data = self._read_files(self.yaml_file_path, deep_merge=deep_merge) + + if self.yaml_config_section is not None: + self.yaml_data = self._traverse_nested_section( + self.yaml_data, self.yaml_config_section, self.yaml_config_section + ) + super().__init__(settings_cls, self.yaml_data) + + def _read_file(self, file_path: Path) -> dict[str, Any]: + import_yaml() + with file_path.open(encoding=self.yaml_file_encoding) as yaml_file: + return yaml.safe_load(yaml_file) or {} + + def _traverse_nested_section( + self, data: dict[str, Any], section_path: str, original_path: str | None = None + ) -> dict[str, Any]: + """ + Traverse nested YAML sections using dot-notation path. + + This method tries to match the longest possible key first before splitting on dots, + allowing access to YAML keys that contain literal dot characters. + + For example, with section_path="a.b.c", it will try: + 1. "a.b.c" as a literal key + 2. "a.b" as a key, then traverse to "c" + 3. "a" as a key, then traverse to "b.c" + 4. "a" as a key, then "b" as a key, then "c" as a key + """ + # Track the original path for error messages + if original_path is None: + original_path = section_path + + # Only reject truly empty paths + if not section_path: + raise ValueError('yaml_config_section cannot be empty') + + # Try the full path as a literal key first (even with leading/trailing/consecutive dots) + try: + return data[section_path] + except KeyError: + pass # Not a literal key, try splitting + except TypeError: + raise TypeError( + f'yaml_config_section path "{original_path}" cannot be traversed in {self.yaml_file_path}. ' + f'An intermediate value is not a dictionary.' + ) + + # If path contains no dots, we already tried it as a literal key above + if '.' not in section_path: + raise KeyError(f'yaml_config_section key "{original_path}" not found in {self.yaml_file_path}') + + # Try progressively shorter prefixes (greedy left-to-right approach) + parts = section_path.split('.') + for i in range(len(parts) - 1, 0, -1): + prefix = '.'.join(parts[:i]) + suffix = '.'.join(parts[i:]) + + if prefix in data: + # Found the prefix as a literal key, now recursively traverse the suffix + try: + return self._traverse_nested_section(data[prefix], suffix, original_path) + except TypeError: + raise TypeError( + f'yaml_config_section path "{original_path}" cannot be traversed in {self.yaml_file_path}. ' + f'An intermediate value is not a dictionary.' + ) + + # If we get here, no match was found + raise KeyError(f'yaml_config_section key "{original_path}" not found in {self.yaml_file_path}') + + def __repr__(self) -> str: + return f'{self.__class__.__name__}(yaml_file={self.yaml_file_path})' + + +__all__ = ['YamlConfigSettingsSource']