| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125 |
- """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 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_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_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})'
|
