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

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

54 statements  

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})'