base.py 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579
  1. """Base classes and core functionality for pydantic-settings sources."""
  2. from __future__ import annotations as _annotations
  3. import json
  4. from abc import ABC, abstractmethod
  5. from collections.abc import Sequence
  6. from dataclasses import asdict, is_dataclass
  7. from pathlib import Path
  8. from typing import TYPE_CHECKING, Any, cast, get_args
  9. from pydantic import AliasChoices, AliasPath, BaseModel, TypeAdapter
  10. from pydantic._internal._typing_extra import ( # type: ignore[attr-defined]
  11. get_origin,
  12. )
  13. from pydantic._internal._utils import deep_update, is_model_class
  14. from pydantic.fields import FieldInfo
  15. from typing_inspection.introspection import is_union_origin
  16. from ..exceptions import SettingsError
  17. from ..utils import _lenient_issubclass
  18. from .types import EnvNoneType, EnvPrefixTarget, ForceDecode, NoDecode, PathType, PydanticModel, _CliSubCommand
  19. from .utils import (
  20. _annotation_is_complex,
  21. _get_alias_names,
  22. _get_field_metadata,
  23. _get_model_fields,
  24. _resolve_type_alias,
  25. _strip_annotated,
  26. _union_is_complex,
  27. )
  28. if TYPE_CHECKING:
  29. from pydantic_settings.main import BaseSettings
  30. def get_subcommand(
  31. model: PydanticModel,
  32. is_required: bool = True,
  33. cli_exit_on_error: bool | None = None,
  34. _suppress_errors: list[SettingsError | SystemExit] | None = None,
  35. ) -> PydanticModel | None:
  36. """
  37. Get the subcommand from a model.
  38. Args:
  39. model: The model to get the subcommand from.
  40. is_required: Determines whether a model must have subcommand set and raises error if not
  41. found. Defaults to `True`.
  42. cli_exit_on_error: Determines whether this function exits with error if no subcommand is found.
  43. Defaults to model_config `cli_exit_on_error` value if set. Otherwise, defaults to `True`.
  44. Returns:
  45. The subcommand model if found, otherwise `None`.
  46. Raises:
  47. SystemExit: When no subcommand is found and is_required=`True` and cli_exit_on_error=`True`
  48. (the default).
  49. SettingsError: When no subcommand is found and is_required=`True` and
  50. cli_exit_on_error=`False`.
  51. """
  52. model_cls = type(model)
  53. if cli_exit_on_error is None and is_model_class(model_cls):
  54. model_default = model_cls.model_config.get('cli_exit_on_error')
  55. if isinstance(model_default, bool):
  56. cli_exit_on_error = model_default
  57. if cli_exit_on_error is None:
  58. cli_exit_on_error = True
  59. subcommands: list[str] = []
  60. for field_name, field_info in _get_model_fields(model_cls).items():
  61. if _CliSubCommand in field_info.metadata:
  62. if getattr(model, field_name) is not None:
  63. return getattr(model, field_name)
  64. subcommands.append(field_name)
  65. if is_required:
  66. error_message = (
  67. f'Error: CLI subcommand is required {{{", ".join(subcommands)}}}'
  68. if subcommands
  69. else 'Error: CLI subcommand is required but no subcommands were found.'
  70. )
  71. err = SystemExit(error_message) if cli_exit_on_error else SettingsError(error_message)
  72. if _suppress_errors is None:
  73. raise err
  74. _suppress_errors.append(err)
  75. return None
  76. class PydanticBaseSettingsSource(ABC):
  77. """
  78. Abstract base class for settings sources, every settings source classes should inherit from it.
  79. """
  80. def __init__(self, settings_cls: type[BaseSettings]):
  81. self.settings_cls = settings_cls
  82. self.config = settings_cls.model_config
  83. self._current_state: dict[str, Any] = {}
  84. self._settings_sources_data: dict[str, dict[str, Any]] = {}
  85. def _set_current_state(self, state: dict[str, Any]) -> None:
  86. """
  87. Record the state of settings from the previous settings sources. This should
  88. be called right before __call__.
  89. """
  90. self._current_state = state
  91. def _set_settings_sources_data(self, states: dict[str, dict[str, Any]]) -> None:
  92. """
  93. Record the state of settings from all previous settings sources. This should
  94. be called right before __call__.
  95. """
  96. self._settings_sources_data = states
  97. @property
  98. def current_state(self) -> dict[str, Any]:
  99. """
  100. The current state of the settings, populated by the previous settings sources.
  101. """
  102. return self._current_state
  103. @property
  104. def settings_sources_data(self) -> dict[str, dict[str, Any]]:
  105. """
  106. The state of all previous settings sources.
  107. """
  108. return self._settings_sources_data
  109. @abstractmethod
  110. def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
  111. """
  112. Gets the value, the key for model creation, and a flag to determine whether value is complex.
  113. This is an abstract method that should be overridden in every settings source classes.
  114. Args:
  115. field: The field.
  116. field_name: The field name.
  117. Returns:
  118. A tuple that contains the value, key and a flag to determine whether value is complex.
  119. """
  120. pass
  121. def field_is_complex(self, field: FieldInfo) -> bool:
  122. """
  123. Checks whether a field is complex, in which case it will attempt to be parsed as JSON.
  124. Args:
  125. field: The field.
  126. Returns:
  127. Whether the field is complex.
  128. """
  129. return _annotation_is_complex(field.annotation, field.metadata)
  130. def prepare_field_value(self, field_name: str, field: FieldInfo, value: Any, value_is_complex: bool) -> Any:
  131. """
  132. Prepares the value of a field.
  133. Args:
  134. field_name: The field name.
  135. field: The field.
  136. value: The value of the field that has to be prepared.
  137. value_is_complex: A flag to determine whether value is complex.
  138. Returns:
  139. The prepared value.
  140. """
  141. if value is not None and (self.field_is_complex(field) or value_is_complex):
  142. return self.decode_complex_value(field_name, field, value)
  143. return value
  144. def decode_complex_value(self, field_name: str, field: FieldInfo, value: Any) -> Any:
  145. """
  146. Decode the value for a complex field
  147. Args:
  148. field_name: The field name.
  149. field: The field.
  150. value: The value of the field that has to be prepared.
  151. Returns:
  152. The decoded value for further preparation
  153. """
  154. if field and (
  155. NoDecode in _get_field_metadata(field)
  156. or (self.config.get('enable_decoding') is False and ForceDecode not in field.metadata)
  157. ):
  158. return value
  159. return json.loads(value)
  160. @abstractmethod
  161. def __call__(self) -> dict[str, Any]:
  162. pass
  163. class ConfigFileSourceMixin(ABC):
  164. def _read_files(self, files: PathType | None, deep_merge: bool = False) -> dict[str, Any]:
  165. if files is None:
  166. return {}
  167. if not isinstance(files, Sequence) or isinstance(files, str):
  168. files = [files]
  169. vars: dict[str, Any] = {}
  170. for file in files:
  171. if isinstance(file, str):
  172. file_path = Path(file)
  173. else:
  174. file_path = file
  175. if isinstance(file_path, Path):
  176. file_path = file_path.expanduser()
  177. if not file_path.is_file():
  178. continue
  179. updating_vars = self._read_file(file_path)
  180. if deep_merge:
  181. vars = deep_update(vars, updating_vars)
  182. else:
  183. vars.update(updating_vars)
  184. return vars
  185. @abstractmethod
  186. def _read_file(self, path: Path) -> dict[str, Any]:
  187. pass
  188. class DefaultSettingsSource(PydanticBaseSettingsSource):
  189. """
  190. Source class for loading default object values.
  191. Args:
  192. settings_cls: The Settings class.
  193. nested_model_default_partial_update: Whether to allow partial updates on nested model default object fields.
  194. Defaults to `False`.
  195. """
  196. def __init__(self, settings_cls: type[BaseSettings], nested_model_default_partial_update: bool | None = None):
  197. super().__init__(settings_cls)
  198. self.defaults: dict[str, Any] = {}
  199. self.nested_model_default_partial_update = (
  200. nested_model_default_partial_update
  201. if nested_model_default_partial_update is not None
  202. else self.config.get('nested_model_default_partial_update', False)
  203. )
  204. if self.nested_model_default_partial_update:
  205. for field_name, field_info in settings_cls.model_fields.items():
  206. alias_names, *_ = _get_alias_names(field_name, field_info)
  207. preferred_alias = alias_names[0]
  208. if is_dataclass(type(field_info.default)):
  209. self.defaults[preferred_alias] = asdict(field_info.default)
  210. elif is_model_class(type(field_info.default)):
  211. self.defaults[preferred_alias] = field_info.default.model_dump()
  212. def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
  213. # Nothing to do here. Only implement the return statement to make mypy happy
  214. return None, '', False
  215. def __call__(self) -> dict[str, Any]:
  216. return self.defaults
  217. def __repr__(self) -> str:
  218. return (
  219. f'{self.__class__.__name__}(nested_model_default_partial_update={self.nested_model_default_partial_update})'
  220. )
  221. class InitSettingsSource(PydanticBaseSettingsSource):
  222. """
  223. Source class for loading values provided during settings class initialization.
  224. """
  225. def __init__(
  226. self,
  227. settings_cls: type[BaseSettings],
  228. init_kwargs: dict[str, Any],
  229. nested_model_default_partial_update: bool | None = None,
  230. ):
  231. self.init_kwargs = {}
  232. init_kwarg_names = set(init_kwargs.keys())
  233. for field_name, field_info in settings_cls.model_fields.items():
  234. alias_names, *_ = _get_alias_names(field_name, field_info)
  235. # When populate_by_name is True, allow using the field name as an input key,
  236. # but normalize to the preferred alias to keep keys consistent across sources.
  237. matchable_names = set(alias_names)
  238. include_name = settings_cls.model_config.get('populate_by_name', False) or settings_cls.model_config.get(
  239. 'validate_by_name', False
  240. )
  241. if include_name:
  242. matchable_names.add(field_name)
  243. init_kwarg_name = init_kwarg_names & matchable_names
  244. if init_kwarg_name:
  245. preferred_alias = alias_names[0] if alias_names else field_name
  246. # Choose provided key deterministically: prefer the first alias in alias_names order;
  247. # fall back to field_name if allowed and provided.
  248. provided_key = next((alias for alias in alias_names if alias in init_kwarg_names), None)
  249. if provided_key is None and include_name and field_name in init_kwarg_names:
  250. provided_key = field_name
  251. # provided_key should not be None here because init_kwarg_name is non-empty
  252. assert provided_key is not None
  253. init_kwarg_names -= init_kwarg_name
  254. self.init_kwargs[preferred_alias] = init_kwargs[provided_key]
  255. # Include any remaining init kwargs (e.g., extras) unchanged
  256. # Note: If populate_by_name is True and the provided key is the field name, but
  257. # no alias exists, we keep it as-is so it can be processed as extra if allowed.
  258. self.init_kwargs.update({key: val for key, val in init_kwargs.items() if key in init_kwarg_names})
  259. super().__init__(settings_cls)
  260. self.nested_model_default_partial_update = (
  261. nested_model_default_partial_update
  262. if nested_model_default_partial_update is not None
  263. else self.config.get('nested_model_default_partial_update', False)
  264. )
  265. def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
  266. # Nothing to do here. Only implement the return statement to make mypy happy
  267. return None, '', False
  268. def __call__(self) -> dict[str, Any]:
  269. return (
  270. TypeAdapter(dict[str, Any]).dump_python(self.init_kwargs)
  271. if self.nested_model_default_partial_update
  272. else self.init_kwargs
  273. )
  274. def __repr__(self) -> str:
  275. return f'{self.__class__.__name__}(init_kwargs={self.init_kwargs!r})'
  276. class PydanticBaseEnvSettingsSource(PydanticBaseSettingsSource):
  277. def __init__(
  278. self,
  279. settings_cls: type[BaseSettings],
  280. case_sensitive: bool | None = None,
  281. env_prefix: str | None = None,
  282. env_prefix_target: EnvPrefixTarget | None = None,
  283. env_ignore_empty: bool | None = None,
  284. env_parse_none_str: str | None = None,
  285. env_parse_enums: bool | None = None,
  286. ) -> None:
  287. super().__init__(settings_cls)
  288. self.case_sensitive = case_sensitive if case_sensitive is not None else self.config.get('case_sensitive', False)
  289. self.env_prefix = env_prefix if env_prefix is not None else self.config.get('env_prefix', '')
  290. self.env_prefix_target = (
  291. env_prefix_target if env_prefix_target is not None else self.config.get('env_prefix_target', 'variable')
  292. )
  293. self.env_ignore_empty = (
  294. env_ignore_empty if env_ignore_empty is not None else self.config.get('env_ignore_empty', False)
  295. )
  296. self.env_parse_none_str = (
  297. env_parse_none_str if env_parse_none_str is not None else self.config.get('env_parse_none_str')
  298. )
  299. self.env_parse_enums = env_parse_enums if env_parse_enums is not None else self.config.get('env_parse_enums')
  300. def _apply_case_sensitive(self, value: str) -> str:
  301. return value.lower() if not self.case_sensitive else value
  302. def _extract_field_info(self, field: FieldInfo, field_name: str) -> list[tuple[str, str, bool]]:
  303. """
  304. Extracts field info. This info is used to get the value of field from environment variables.
  305. It returns a list of tuples, each tuple contains:
  306. * field_key: The key of field that has to be used in model creation.
  307. * env_name: The environment variable name of the field.
  308. * value_is_complex: A flag to determine whether the value from environment variable
  309. is complex and has to be parsed.
  310. Args:
  311. field (FieldInfo): The field.
  312. field_name (str): The field name.
  313. Returns:
  314. list[tuple[str, str, bool]]: List of tuples, each tuple contains field_key, env_name, and value_is_complex.
  315. """
  316. field_info: list[tuple[str, str, bool]] = []
  317. if isinstance(field.validation_alias, (AliasChoices, AliasPath)):
  318. v_alias: str | list[str | int] | list[list[str | int]] | None = field.validation_alias.convert_to_aliases()
  319. else:
  320. v_alias = field.validation_alias
  321. if v_alias:
  322. env_prefix = self.env_prefix if self.env_prefix_target in ('alias', 'all') else ''
  323. if isinstance(v_alias, list): # AliasChoices, AliasPath
  324. for alias in v_alias:
  325. if isinstance(alias, str): # AliasPath
  326. field_info.append(
  327. (alias, self._apply_case_sensitive(env_prefix + alias), True if len(alias) > 1 else False)
  328. )
  329. elif isinstance(alias, list): # AliasChoices
  330. first_arg = cast(str, alias[0]) # first item of an AliasChoices must be a str
  331. field_info.append(
  332. (
  333. first_arg,
  334. self._apply_case_sensitive(env_prefix + first_arg),
  335. True if len(alias) > 1 else False,
  336. )
  337. )
  338. else: # string validation alias
  339. field_info.append((v_alias, self._apply_case_sensitive(env_prefix + v_alias), False))
  340. if not v_alias or self.config.get('populate_by_name', False) or self.config.get('validate_by_name', False):
  341. annotation = _strip_annotated(_resolve_type_alias(field.annotation))
  342. env_prefix = self.env_prefix if self.env_prefix_target in ('variable', 'all') else ''
  343. if is_union_origin(get_origin(annotation)) and _union_is_complex(annotation, field.metadata):
  344. field_info.append((field_name, self._apply_case_sensitive(env_prefix + field_name), True))
  345. else:
  346. field_info.append((field_name, self._apply_case_sensitive(env_prefix + field_name), False))
  347. return field_info
  348. def _replace_field_names_case_insensitively(self, field: FieldInfo, field_values: dict[str, Any]) -> dict[str, Any]:
  349. """
  350. Replace field names in values dict by looking in models fields insensitively.
  351. By having the following models:
  352. ```py
  353. class SubSubSub(BaseModel):
  354. VaL3: str
  355. class SubSub(BaseModel):
  356. Val2: str
  357. SUB_sub_SuB: SubSubSub
  358. class Sub(BaseModel):
  359. VAL1: str
  360. SUB_sub: SubSub
  361. class Settings(BaseSettings):
  362. nested: Sub
  363. model_config = SettingsConfigDict(env_nested_delimiter='__')
  364. ```
  365. Then:
  366. _replace_field_names_case_insensitively(
  367. field,
  368. {"val1": "v1", "sub_SUB": {"VAL2": "v2", "sub_SUB_sUb": {"vAl3": "v3"}}}
  369. )
  370. Returns {'VAL1': 'v1', 'SUB_sub': {'Val2': 'v2', 'SUB_sub_SuB': {'VaL3': 'v3'}}}
  371. """
  372. values: dict[str, Any] = {}
  373. for name, value in field_values.items():
  374. sub_model_field: FieldInfo | None = None
  375. annotation = field.annotation
  376. # If field is Optional, we need to find the actual type
  377. if is_union_origin(get_origin(field.annotation)):
  378. args = get_args(annotation)
  379. if len(args) == 2 and type(None) in args:
  380. for arg in args:
  381. if arg is not None:
  382. annotation = arg
  383. break
  384. # This is here to make mypy happy
  385. # Item "None" of "Optional[Type[Any]]" has no attribute "model_fields"
  386. if not annotation or not hasattr(annotation, 'model_fields'):
  387. values[name] = value
  388. continue
  389. else:
  390. model_fields: dict[str, FieldInfo] = annotation.model_fields
  391. # Find field in sub model by looking in fields case insensitively
  392. field_key: str | None = None
  393. for sub_model_field_name, sub_model_field in model_fields.items():
  394. aliases, _ = _get_alias_names(sub_model_field_name, sub_model_field)
  395. _search = (alias for alias in aliases if alias.lower() == name.lower())
  396. if field_key := next(_search, None):
  397. break
  398. if not field_key:
  399. values[name] = value
  400. continue
  401. if (
  402. sub_model_field is not None
  403. and _lenient_issubclass(sub_model_field.annotation, BaseModel)
  404. and isinstance(value, dict)
  405. ):
  406. values[field_key] = self._replace_field_names_case_insensitively(sub_model_field, value)
  407. else:
  408. values[field_key] = value
  409. return values
  410. def _replace_env_none_type_values(self, field_value: dict[str, Any]) -> dict[str, Any]:
  411. """
  412. Recursively parse values that are of "None" type(EnvNoneType) to `None` type(None).
  413. """
  414. values: dict[str, Any] = {}
  415. for key, value in field_value.items():
  416. if not isinstance(value, EnvNoneType):
  417. values[key] = value if not isinstance(value, dict) else self._replace_env_none_type_values(value)
  418. else:
  419. values[key] = None
  420. return values
  421. def _get_resolved_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
  422. """
  423. Gets the value, the preferred alias key for model creation, and a flag to determine whether value
  424. is complex.
  425. Note:
  426. In V3, this method should either be made public, or, this method should be removed and the
  427. abstract method get_field_value should be updated to include a "use_preferred_alias" flag.
  428. Args:
  429. field: The field.
  430. field_name: The field name.
  431. Returns:
  432. A tuple that contains the value, preferred key and a flag to determine whether value is complex.
  433. """
  434. field_value, field_key, value_is_complex = self.get_field_value(field, field_name)
  435. # Only use preferred_key when no value was found; otherwise preserve the key that matched
  436. if field_value is None and not (
  437. value_is_complex
  438. or (
  439. (self.config.get('populate_by_name', False) or self.config.get('validate_by_name', False))
  440. and (field_key == field_name)
  441. )
  442. ):
  443. field_infos = self._extract_field_info(field, field_name)
  444. preferred_key, *_ = field_infos[0]
  445. return field_value, preferred_key, value_is_complex
  446. return field_value, field_key, value_is_complex
  447. def __call__(self) -> dict[str, Any]:
  448. data: dict[str, Any] = {}
  449. for field_name, field in self.settings_cls.model_fields.items():
  450. try:
  451. field_value, field_key, value_is_complex = self._get_resolved_field_value(field, field_name)
  452. except Exception as e:
  453. raise SettingsError(
  454. f'error getting value for field "{field_name}" from source "{self.__class__.__name__}"'
  455. ) from e
  456. try:
  457. field_value = self.prepare_field_value(field_name, field, field_value, value_is_complex)
  458. except ValueError as e:
  459. raise SettingsError(
  460. f'error parsing value for field "{field_name}" from source "{self.__class__.__name__}"'
  461. ) from e
  462. if field_value is not None:
  463. if self.env_parse_none_str is not None:
  464. if isinstance(field_value, dict):
  465. field_value = self._replace_env_none_type_values(field_value)
  466. elif isinstance(field_value, EnvNoneType):
  467. field_value = None
  468. if (
  469. not self.case_sensitive
  470. # and _lenient_issubclass(field.annotation, BaseModel)
  471. and isinstance(field_value, dict)
  472. ):
  473. data[field_key] = self._replace_field_names_case_insensitively(field, field_value)
  474. else:
  475. data[field_key] = field_value
  476. return data
  477. __all__ = [
  478. 'ConfigFileSourceMixin',
  479. 'DefaultSettingsSource',
  480. 'InitSettingsSource',
  481. 'PydanticBaseEnvSettingsSource',
  482. 'PydanticBaseSettingsSource',
  483. 'SettingsError',
  484. ]