1"""Secrets file settings source."""
2
3from __future__ import annotations as _annotations
4
5import os
6import warnings
7from pathlib import Path
8from typing import (
9 TYPE_CHECKING,
10 Any,
11)
12
13from pydantic.fields import FieldInfo
14
15from pydantic_settings.utils import path_type_label
16
17from ...exceptions import SettingsError
18from ..base import PydanticBaseEnvSettingsSource
19from ..types import PathType
20
21if TYPE_CHECKING:
22 from pydantic_settings.main import BaseSettings
23
24
25class SecretsSettingsSource(PydanticBaseEnvSettingsSource):
26 """
27 Source class for loading settings values from secret files.
28 """
29
30 def __init__(
31 self,
32 settings_cls: type[BaseSettings],
33 secrets_dir: PathType | None = None,
34 case_sensitive: bool | None = None,
35 env_prefix: str | None = None,
36 env_ignore_empty: bool | None = None,
37 env_parse_none_str: str | None = None,
38 env_parse_enums: bool | None = None,
39 ) -> None:
40 super().__init__(
41 settings_cls, case_sensitive, env_prefix, env_ignore_empty, env_parse_none_str, env_parse_enums
42 )
43 self.secrets_dir = secrets_dir if secrets_dir is not None else self.config.get('secrets_dir')
44
45 def __call__(self) -> dict[str, Any]:
46 """
47 Build fields from "secrets" files.
48 """
49 secrets: dict[str, str | None] = {}
50
51 if self.secrets_dir is None:
52 return secrets
53
54 secrets_dirs = [self.secrets_dir] if isinstance(self.secrets_dir, (str, os.PathLike)) else self.secrets_dir
55 secrets_paths = [Path(p).expanduser() for p in secrets_dirs]
56 self.secrets_paths = []
57
58 for path in secrets_paths:
59 if not path.exists():
60 warnings.warn(f'directory "{path}" does not exist')
61 else:
62 self.secrets_paths.append(path)
63
64 if not len(self.secrets_paths):
65 return secrets
66
67 for path in self.secrets_paths:
68 if not path.is_dir():
69 raise SettingsError(f'secrets_dir must reference a directory, not a {path_type_label(path)}')
70
71 return super().__call__()
72
73 @classmethod
74 def find_case_path(cls, dir_path: Path, file_name: str, case_sensitive: bool) -> Path | None:
75 """
76 Find a file within path's directory matching filename, optionally ignoring case.
77
78 Args:
79 dir_path: Directory path.
80 file_name: File name.
81 case_sensitive: Whether to search for file name case sensitively.
82
83 Returns:
84 Whether file path or `None` if file does not exist in directory.
85 """
86 for f in dir_path.iterdir():
87 if f.name == file_name:
88 return f
89 elif not case_sensitive and f.name.lower() == file_name.lower():
90 return f
91 return None
92
93 def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
94 """
95 Gets the value for field from secret file and a flag to determine whether value is complex.
96
97 Args:
98 field: The field.
99 field_name: The field name.
100
101 Returns:
102 A tuple that contains the value (`None` if the file does not exist), key, and
103 a flag to determine whether value is complex.
104 """
105
106 for field_key, env_name, value_is_complex in self._extract_field_info(field, field_name):
107 # paths reversed to match the last-wins behaviour of `env_file`
108 for secrets_path in reversed(self.secrets_paths):
109 path = self.find_case_path(secrets_path, env_name, self.case_sensitive)
110 if not path:
111 # path does not exist, we currently don't return a warning for this
112 continue
113
114 if path.is_file():
115 return path.read_text().strip(), field_key, value_is_complex
116 else:
117 warnings.warn(
118 f'attempted to load secret file "{path}" but found a {path_type_label(path)} instead.',
119 stacklevel=4,
120 )
121
122 return None, field_key, value_is_complex
123
124 def __repr__(self) -> str:
125 return f'{self.__class__.__name__}(secrets_dir={self.secrets_dir!r})'