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 EnvPrefixTarget, 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_prefix_target: EnvPrefixTarget | None = None,
37 env_ignore_empty: bool | None = None,
38 env_parse_none_str: str | None = None,
39 env_parse_enums: bool | None = None,
40 ) -> None:
41 super().__init__(
42 settings_cls,
43 case_sensitive,
44 env_prefix,
45 env_prefix_target,
46 env_ignore_empty,
47 env_parse_none_str,
48 env_parse_enums,
49 )
50 self.secrets_dir = secrets_dir if secrets_dir is not None else self.config.get('secrets_dir')
51
52 def __call__(self) -> dict[str, Any]:
53 """
54 Build fields from "secrets" files.
55 """
56 secrets: dict[str, str | None] = {}
57
58 if self.secrets_dir is None:
59 return secrets
60
61 secrets_dirs = [self.secrets_dir] if isinstance(self.secrets_dir, (str, os.PathLike)) else self.secrets_dir
62 secrets_paths = [Path(p).expanduser() for p in secrets_dirs]
63 self.secrets_paths = []
64
65 for path in secrets_paths:
66 if not path.exists():
67 warnings.warn(f'directory "{path}" does not exist')
68 else:
69 self.secrets_paths.append(path)
70
71 if not len(self.secrets_paths):
72 return secrets
73
74 for path in self.secrets_paths:
75 if not path.is_dir():
76 raise SettingsError(f'secrets_dir must reference a directory, not a {path_type_label(path)}')
77
78 return super().__call__()
79
80 @classmethod
81 def find_case_path(cls, dir_path: Path, file_name: str, case_sensitive: bool) -> Path | None:
82 """
83 Find a file within path's directory matching filename, optionally ignoring case.
84
85 Args:
86 dir_path: Directory path.
87 file_name: File name.
88 case_sensitive: Whether to search for file name case sensitively.
89
90 Returns:
91 Whether file path or `None` if file does not exist in directory.
92 """
93 for f in dir_path.iterdir():
94 if f.name == file_name:
95 return f
96 elif not case_sensitive and f.name.lower() == file_name.lower():
97 return f
98 return None
99
100 def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
101 """
102 Gets the value for field from secret file and a flag to determine whether value is complex.
103
104 Args:
105 field: The field.
106 field_name: The field name.
107
108 Returns:
109 A tuple that contains the value (`None` if the file does not exist), key, and
110 a flag to determine whether value is complex.
111 """
112
113 for field_key, env_name, value_is_complex in self._extract_field_info(field, field_name):
114 # paths reversed to match the last-wins behaviour of `env_file`
115 for secrets_path in reversed(self.secrets_paths):
116 path = self.find_case_path(secrets_path, env_name, self.case_sensitive)
117 if not path:
118 # path does not exist, we currently don't return a warning for this
119 continue
120
121 if path.is_file():
122 return path.read_text().strip(), field_key, value_is_complex
123 else:
124 warnings.warn(
125 f'attempted to load secret file "{path}" but found a {path_type_label(path)} instead.',
126 stacklevel=4,
127 )
128
129 return None, field_key, value_is_complex
130
131 def __repr__(self) -> str:
132 return f'{self.__class__.__name__}(secrets_dir={self.secrets_dir!r})'