Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pydantic_settings/sources/providers/secrets.py: 17%

1"""Secrets file settings source.""" 

2 

3from __future__ import annotations as _annotations 

4 

5import logging 

6import os 

7import warnings 

8from pathlib import Path 

9from typing import ( 

10 TYPE_CHECKING, 

11 Any, 

12) 

13 

14from pydantic.fields import FieldInfo 

15 

16from pydantic_settings.utils import _settings_debug_enabled, logger, path_type_label 

17 

18from ...exceptions import SettingsError 

19from ..base import PydanticBaseEnvSettingsSource 

20from ..types import EnvPrefixTarget, PathType 

21from ..utils import InitState 

22 

23if TYPE_CHECKING: 

24 from pydantic_settings.main import BaseSettings 

25 

26 

27class SecretsSettingsSource(PydanticBaseEnvSettingsSource): 

28 """ 

29 Source class for loading settings values from secret files. 

30 """ 

31 

32 def __init__( 

33 self, 

34 settings_cls: type[BaseSettings], 

35 secrets_dir: PathType | None = None, 

36 case_sensitive: bool | None = None, 

37 env_prefix: str | None = None, 

38 env_prefix_target: EnvPrefixTarget | None = None, 

39 env_ignore_empty: bool | None = None, 

40 env_parse_none_str: str | None = None, 

41 env_parse_enums: bool | None = None, 

42 _init_state: InitState | None = None, 

43 ) -> None: 

44 super().__init__( 

45 settings_cls, 

46 case_sensitive, 

47 env_prefix, 

48 env_prefix_target, 

49 env_ignore_empty, 

50 env_parse_none_str, 

51 env_parse_enums, 

52 _init_state, 

53 ) 

54 self.secrets_dir = secrets_dir if secrets_dir is not None else self.config.get('secrets_dir') 

55 

56 def __call__(self) -> dict[str, Any]: 

57 """ 

58 Build fields from "secrets" files. 

59 """ 

60 secrets: dict[str, str | None] = {} 

61 

62 if self.secrets_dir is None: 

63 return secrets 

64 

65 secrets_dirs = [self.secrets_dir] if isinstance(self.secrets_dir, (str, os.PathLike)) else self.secrets_dir 

66 secrets_paths = [Path(p).expanduser() for p in secrets_dirs] 

67 self.secrets_paths = [] 

68 

69 for path in secrets_paths: 

70 if not path.exists(): 

71 # No `stacklevel`: this source is reached at different depths — via 

72 # `Settings()` (through `_settings_build_values`) or by instantiating 

73 # the source directly — so no fixed value attributes the warning to 

74 # user code in both cases. 

75 warnings.warn(f'directory "{path}" does not exist') # noqa: B028 

76 else: 

77 self.secrets_paths.append(path) 

78 

79 if not len(self.secrets_paths): 

80 return secrets 

81 

82 for path in self.secrets_paths: 

83 if not path.is_dir(): 

84 raise SettingsError(f'secrets_dir must reference a directory, not a {path_type_label(path)}') 

85 

86 return super().__call__() 

87 

88 @classmethod 

89 def find_case_path(cls, dir_path: Path, file_name: str, case_sensitive: bool) -> Path | None: 

90 """ 

91 Find a file within path's directory matching filename, optionally ignoring case. 

92 

93 Args: 

94 dir_path: Directory path. 

95 file_name: File name. 

96 case_sensitive: Whether to search for file name case sensitively. 

97 

98 Returns: 

99 Whether file path or `None` if file does not exist in directory. 

100 """ 

101 for f in dir_path.iterdir(): 

102 if f.name == file_name or (not case_sensitive and f.name.lower() == file_name.lower()): 

103 return f 

104 return None 

105 

106 def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]: 

107 """ 

108 Gets the value for field from secret file and a flag to determine whether value is complex. 

109 

110 Args: 

111 field: The field. 

112 field_name: The field name. 

113 

114 Returns: 

115 A tuple that contains the value (`None` if the file does not exist), key, and 

116 a flag to determine whether value is complex. 

117 """ 

118 

119 debug = _settings_debug_enabled() and logger.isEnabledFor(logging.DEBUG) 

120 

121 for field_key, env_name, value_is_complex in self._extract_field_info(field, field_name): 

122 # paths reversed to match the last-wins behaviour of `env_file` 

123 for secrets_path in reversed(self.secrets_paths): 

124 path = self.find_case_path(secrets_path, env_name, self.case_sensitive) 

125 if not path: 

126 # path does not exist, we currently don't return a warning for this 

127 if debug: 

128 logger.debug('Secret file not found, skipping: %s', (secrets_path / env_name).resolve()) 

129 continue 

130 

131 if path.is_file(): 

132 if debug: 

133 logger.debug('Loading secret file: %s', path.resolve()) 

134 # Explicit encoding: `read_text()` would otherwise decode with the 

135 # locale encoding, so a UTF-8 secret is corrupted on a non-UTF-8 

136 # Windows code page (silently, when every byte happens to map). 

137 return path.read_text(encoding='utf-8').strip(), field_key, value_is_complex 

138 warnings.warn( 

139 f'attempted to load secret file "{path}" but found a {path_type_label(path)} instead.', 

140 stacklevel=4, 

141 ) 

142 

143 return None, field_key, value_is_complex 

144 

145 def __repr__(self) -> str: 

146 return f'{self.__class__.__name__}(secrets_dir={self.secrets_dir!r})'