Coverage for pydantic/env_settings.py: 100.00%

190 statements  

« prev     ^ index     » next       coverage.py v7.6.1, created at 2024-08-15 13:26 +0000

1import os 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

2import warnings 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

3from pathlib import Path 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

4from typing import AbstractSet, Any, Callable, ClassVar, Dict, List, Mapping, Optional, Tuple, Type, Union 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

5 

6from pydantic.config import BaseConfig, Extra 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

7from pydantic.fields import ModelField 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

8from pydantic.main import BaseModel 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

9from pydantic.types import JsonWrapper 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

10from pydantic.typing import StrPath, display_as_type, get_origin, is_union 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

11from pydantic.utils import deep_update, lenient_issubclass, path_type, sequence_like 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

12 

13env_file_sentinel = str(object()) 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

14 

15SettingsSourceCallable = Callable[['BaseSettings'], Dict[str, Any]] 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

16DotenvType = Union[StrPath, List[StrPath], Tuple[StrPath, ...]] 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

17 

18 

19class SettingsError(ValueError): 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

20 pass 1EMaubvcwdxeyLOpGqHrIsJtKPQRSTUFNfzgAhBiCjD

21 

22 

23class BaseSettings(BaseModel): 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

24 """ 

25 Base class for settings, allowing values to be overridden by environment variables. 

26 

27 This is useful in production for secrets you do not wish to save in code, it plays nicely with docker(-compose), 

28 Heroku and any 12 factor app design. 

29 """ 

30 

31 def __init__( 1aubvcwdxeypGqHrIsJtKklmnPQRSTUofzgAhBiCjD

32 __pydantic_self__, 

33 _env_file: Optional[DotenvType] = env_file_sentinel, 

34 _env_file_encoding: Optional[str] = None, 

35 _env_nested_delimiter: Optional[str] = None, 

36 _secrets_dir: Optional[StrPath] = None, 

37 **values: Any, 

38 ) -> None: 

39 # Uses something other than `self` the first arg to allow "self" as a settable attribute 

40 super().__init__( 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

41 **__pydantic_self__._build_values( 

42 values, 

43 _env_file=_env_file, 

44 _env_file_encoding=_env_file_encoding, 

45 _env_nested_delimiter=_env_nested_delimiter, 

46 _secrets_dir=_secrets_dir, 

47 ) 

48 ) 

49 

50 def _build_values( 1aubvcwdxeypGqHrIsJtKklmnPQRSTUofzgAhBiCjD

51 self, 

52 init_kwargs: Dict[str, Any], 

53 _env_file: Optional[DotenvType] = None, 

54 _env_file_encoding: Optional[str] = None, 

55 _env_nested_delimiter: Optional[str] = None, 

56 _secrets_dir: Optional[StrPath] = None, 

57 ) -> Dict[str, Any]: 

58 # Configure built-in sources 

59 init_settings = InitSettingsSource(init_kwargs=init_kwargs) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

60 env_settings = EnvSettingsSource( 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

61 env_file=(_env_file if _env_file != env_file_sentinel else self.__config__.env_file), 

62 env_file_encoding=( 

63 _env_file_encoding if _env_file_encoding is not None else self.__config__.env_file_encoding 

64 ), 

65 env_nested_delimiter=( 

66 _env_nested_delimiter if _env_nested_delimiter is not None else self.__config__.env_nested_delimiter 

67 ), 

68 env_prefix_len=len(self.__config__.env_prefix), 

69 ) 

70 file_secret_settings = SecretsSettingsSource(secrets_dir=_secrets_dir or self.__config__.secrets_dir) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

71 # Provide a hook to set built-in sources priority and add / remove sources 

72 sources = self.__config__.customise_sources( 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

73 init_settings=init_settings, env_settings=env_settings, file_secret_settings=file_secret_settings 

74 ) 

75 if sources: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

76 return deep_update(*reversed([source(self) for source in sources])) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

77 else: 

78 # no one should mean to do this, but I think returning an empty dict is marginally preferable 

79 # to an informative error and much better than a confusing error 

80 return {} 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

81 

82 class Config(BaseConfig): 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

83 env_prefix: str = '' 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

84 env_file: Optional[DotenvType] = None 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

85 env_file_encoding: Optional[str] = None 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

86 env_nested_delimiter: Optional[str] = None 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

87 secrets_dir: Optional[StrPath] = None 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

88 validate_all: bool = True 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

89 extra: Extra = Extra.forbid 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

90 arbitrary_types_allowed: bool = True 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

91 case_sensitive: bool = False 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

92 

93 @classmethod 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

94 def prepare_field(cls, field: ModelField) -> None: 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

95 env_names: Union[List[str], AbstractSet[str]] 

96 field_info_from_config = cls.get_field_info(field.name) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

97 

98 env = field_info_from_config.get('env') or field.field_info.extra.get('env') 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

99 if env is None: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

100 if field.has_alias: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

101 warnings.warn( 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

102 'aliases are no longer used by BaseSettings to define which environment variables to read. ' 

103 'Instead use the "env" field setting. ' 

104 'See https://pydantic-docs.helpmanual.io/usage/settings/#environment-variable-names', 

105 FutureWarning, 

106 ) 

107 env_names = {cls.env_prefix + field.name} 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

108 elif isinstance(env, str): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

109 env_names = {env} 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

110 elif isinstance(env, (set, frozenset)): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

111 env_names = env 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

112 elif sequence_like(env): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

113 env_names = list(env) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

114 else: 

115 raise TypeError(f'invalid field env: {env!r} ({display_as_type(env)}); should be string, list or set') 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

116 

117 if not cls.case_sensitive: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

118 env_names = env_names.__class__(n.lower() for n in env_names) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

119 field.field_info.extra['env_names'] = env_names 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

120 

121 @classmethod 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

122 def customise_sources( 1aubvcwdxeypGqHrIsJtKklmnPQRSTUofzgAhBiCjD

123 cls, 

124 init_settings: SettingsSourceCallable, 

125 env_settings: SettingsSourceCallable, 

126 file_secret_settings: SettingsSourceCallable, 

127 ) -> Tuple[SettingsSourceCallable, ...]: 

128 return init_settings, env_settings, file_secret_settings 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

129 

130 @classmethod 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

131 def parse_env_var(cls, field_name: str, raw_val: str) -> Any: 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

132 return cls.json_loads(raw_val) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

133 

134 # populated by the metaclass using the Config class defined above, annotated here to help IDEs only 

135 __config__: ClassVar[Type[Config]] 1EMaubvcwdxeyLOpGqHrIsJtKPQRSTUFNfzgAhBiCjD

136 

137 

138class InitSettingsSource: 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

139 __slots__ = ('init_kwargs',) 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

140 

141 def __init__(self, init_kwargs: Dict[str, Any]): 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

142 self.init_kwargs = init_kwargs 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

143 

144 def __call__(self, settings: BaseSettings) -> Dict[str, Any]: 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

145 return self.init_kwargs 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

146 

147 def __repr__(self) -> str: 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

148 return f'InitSettingsSource(init_kwargs={self.init_kwargs!r})' 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

149 

150 

151class EnvSettingsSource: 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

152 __slots__ = ('env_file', 'env_file_encoding', 'env_nested_delimiter', 'env_prefix_len') 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

153 

154 def __init__( 1aubvcwdxeypGqHrIsJtKklmnPQRSTUofzgAhBiCjD

155 self, 

156 env_file: Optional[DotenvType], 

157 env_file_encoding: Optional[str], 

158 env_nested_delimiter: Optional[str] = None, 

159 env_prefix_len: int = 0, 

160 ): 

161 self.env_file: Optional[DotenvType] = env_file 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

162 self.env_file_encoding: Optional[str] = env_file_encoding 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

163 self.env_nested_delimiter: Optional[str] = env_nested_delimiter 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

164 self.env_prefix_len: int = env_prefix_len 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

165 

166 def __call__(self, settings: BaseSettings) -> Dict[str, Any]: # noqa C901 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

167 """ 

168 Build environment variables suitable for passing to the Model. 

169 """ 

170 d: Dict[str, Any] = {} 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

171 

172 if settings.__config__.case_sensitive: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

173 env_vars: Mapping[str, Optional[str]] = os.environ 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

174 else: 

175 env_vars = {k.lower(): v for k, v in os.environ.items()} 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

176 

177 dotenv_vars = self._read_env_files(settings.__config__.case_sensitive) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

178 if dotenv_vars: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

179 env_vars = {**dotenv_vars, **env_vars} 1EabcdeLpqrstklmnoFfghij

180 

181 for field in settings.__fields__.values(): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

182 env_val: Optional[str] = None 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

183 for env_name in field.field_info.extra['env_names']: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

184 env_val = env_vars.get(env_name) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

185 if env_val is not None: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

186 break 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

187 

188 is_complex, allow_parse_failure = self.field_is_complex(field) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

189 if is_complex: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

190 if env_val is None: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

191 # field is complex but no value found so far, try explode_env_vars 

192 env_val_built = self.explode_env_vars(field, env_vars) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

193 if env_val_built: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

194 d[field.alias] = env_val_built 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

195 else: 

196 # field is complex and there's a value, decode that as JSON, then add explode_env_vars 

197 try: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

198 env_val = settings.__config__.parse_env_var(field.name, env_val) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

199 except ValueError as e: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

200 if not allow_parse_failure: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

201 raise SettingsError(f'error parsing env var "{env_name}"') from e 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

202 

203 if isinstance(env_val, dict): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

204 d[field.alias] = deep_update(env_val, self.explode_env_vars(field, env_vars)) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

205 else: 

206 d[field.alias] = env_val 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

207 elif env_val is not None: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

208 # simplest case, field is not complex, we only need to add the value if it was found 

209 d[field.alias] = env_val 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

210 

211 return d 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

212 

213 def _read_env_files(self, case_sensitive: bool) -> Dict[str, Optional[str]]: 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

214 env_files = self.env_file 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

215 if env_files is None: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

216 return {} 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

217 

218 if isinstance(env_files, (str, os.PathLike)): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

219 env_files = [env_files] 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

220 

221 dotenv_vars = {} 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

222 for env_file in env_files: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

223 env_path = Path(env_file).expanduser() 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

224 if env_path.is_file(): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

225 dotenv_vars.update( 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

226 read_env_file(env_path, encoding=self.env_file_encoding, case_sensitive=case_sensitive) 

227 ) 

228 

229 return dotenv_vars 1EabcdeLpqrstklmnoFfghij

230 

231 def field_is_complex(self, field: ModelField) -> Tuple[bool, bool]: 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

232 """ 

233 Find out if a field is complex, and if so whether JSON errors should be ignored 

234 """ 

235 if lenient_issubclass(field.annotation, JsonWrapper): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

236 return False, False 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

237 

238 if field.is_complex(): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

239 allow_parse_failure = False 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

240 elif is_union(get_origin(field.type_)) and field.sub_fields and any(f.is_complex() for f in field.sub_fields): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

241 allow_parse_failure = True 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

242 else: 

243 return False, False 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

244 

245 return True, allow_parse_failure 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

246 

247 def explode_env_vars(self, field: ModelField, env_vars: Mapping[str, Optional[str]]) -> Dict[str, Any]: 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

248 """ 

249 Process env_vars and extract the values of keys containing env_nested_delimiter into nested dictionaries. 

250 

251 This is applied to a single field, hence filtering by env_var prefix. 

252 """ 

253 prefixes = [f'{env_name}{self.env_nested_delimiter}' for env_name in field.field_info.extra['env_names']] 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

254 result: Dict[str, Any] = {} 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

255 for env_name, env_val in env_vars.items(): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

256 if not any(env_name.startswith(prefix) for prefix in prefixes): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

257 continue 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

258 # we remove the prefix before splitting in case the prefix has characters in common with the delimiter 

259 env_name_without_prefix = env_name[self.env_prefix_len :] 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

260 _, *keys, last_key = env_name_without_prefix.split(self.env_nested_delimiter) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

261 env_var = result 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

262 for key in keys: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

263 env_var = env_var.setdefault(key, {}) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

264 env_var[last_key] = env_val 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

265 

266 return result 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

267 

268 def __repr__(self) -> str: 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

269 return ( 1aubvcwdxeypGqHrIsJtKklmnofzgAhBiCjD

270 f'EnvSettingsSource(env_file={self.env_file!r}, env_file_encoding={self.env_file_encoding!r}, ' 

271 f'env_nested_delimiter={self.env_nested_delimiter!r})' 

272 ) 

273 

274 

275class SecretsSettingsSource: 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

276 __slots__ = ('secrets_dir',) 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

277 

278 def __init__(self, secrets_dir: Optional[StrPath]): 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

279 self.secrets_dir: Optional[StrPath] = secrets_dir 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

280 

281 def __call__(self, settings: BaseSettings) -> Dict[str, Any]: 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

282 """ 

283 Build fields from "secrets" files. 

284 """ 

285 secrets: Dict[str, Optional[str]] = {} 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

286 

287 if self.secrets_dir is None: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

288 return secrets 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

289 

290 secrets_path = Path(self.secrets_dir).expanduser() 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

291 

292 if not secrets_path.exists(): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

293 warnings.warn(f'directory "{secrets_path}" does not exist') 1EMaubvcwdxeyklmnoFNfzgAhBiCjD

294 return secrets 1EMaubvcwdxeyklmnoFNfzgAhBiCjD

295 

296 if not secrets_path.is_dir(): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

297 raise SettingsError(f'secrets_dir must reference a directory, not a {path_type(secrets_path)}') 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

298 

299 for field in settings.__fields__.values(): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

300 for env_name in field.field_info.extra['env_names']: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

301 path = find_case_path(secrets_path, env_name, settings.__config__.case_sensitive) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

302 if not path: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

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

304 continue 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

305 

306 if path.is_file(): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

307 secret_value = path.read_text().strip() 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

308 if field.is_complex(): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

309 try: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

310 secret_value = settings.__config__.parse_env_var(field.name, secret_value) 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

311 except ValueError as e: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

312 raise SettingsError(f'error parsing env var "{env_name}"') from e 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

313 

314 secrets[field.alias] = secret_value 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

315 else: 

316 warnings.warn( 1EMaubvcwdxeyklmnoFNfzgAhBiCjD

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

318 stacklevel=4, 

319 ) 

320 return secrets 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

321 

322 def __repr__(self) -> str: 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

323 return f'SecretsSettingsSource(secrets_dir={self.secrets_dir!r})' 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

324 

325 

326def read_env_file( 1aubvcwdxeypGqHrIsJtKklmnPQRSTUofzgAhBiCjD

327 file_path: StrPath, *, encoding: str = None, case_sensitive: bool = False 

328) -> Dict[str, Optional[str]]: 

329 try: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

330 from dotenv import dotenv_values 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

331 except ImportError as e: 1MuvwxyOGHIJKNzABCD

332 raise ImportError('python-dotenv is not installed, run `pip install pydantic[dotenv]`') from e 1MuvwxyOGHIJKNzABCD

333 

334 file_vars: Dict[str, Optional[str]] = dotenv_values(file_path, encoding=encoding or 'utf8') 1EabcdeLpqrstklmnoFfghij

335 if not case_sensitive: 1EabcdeLpqrstklmnoFfghij

336 return {k.lower(): v for k, v in file_vars.items()} 1EabcdeLpqrstklmnoFfghij

337 else: 

338 return file_vars 1EabcdeLpqrstklmnoFfghij

339 

340 

341def find_case_path(dir_path: Path, file_name: str, case_sensitive: bool) -> Optional[Path]: 1EMaubvcwdxeyLOpGqHrIsJtKklmnPQRSTUoFNfzgAhBiCjD

342 """ 

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

344 """ 

345 for f in dir_path.iterdir(): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

346 if f.name == file_name: 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

347 return f 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

348 elif not case_sensitive and f.name.lower() == file_name.lower(): 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

349 return f 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD

350 return None 1EMaubvcwdxeyLOpGqHrIsJtKklmnoFNfzgAhBiCjD