main.py 42 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901
  1. from __future__ import annotations as _annotations
  2. import asyncio
  3. import inspect
  4. import re
  5. import threading
  6. import warnings
  7. from argparse import Namespace
  8. from collections.abc import Mapping
  9. from types import SimpleNamespace
  10. from typing import Any, ClassVar, Literal, TextIO, TypeVar, cast
  11. from pydantic import ConfigDict
  12. from pydantic._internal._config import config_keys
  13. from pydantic._internal._signature import _field_name_for_signature
  14. from pydantic._internal._utils import deep_update, is_model_class
  15. from pydantic.dataclasses import is_pydantic_dataclass
  16. from pydantic.main import BaseModel
  17. from .exceptions import SettingsError
  18. from .sources import (
  19. ENV_FILE_SENTINEL,
  20. CliSettingsSource,
  21. DefaultSettingsSource,
  22. DotEnvSettingsSource,
  23. DotenvType,
  24. EnvPrefixTarget,
  25. EnvSettingsSource,
  26. InitSettingsSource,
  27. JsonConfigSettingsSource,
  28. PathType,
  29. PydanticBaseSettingsSource,
  30. PydanticModel,
  31. PyprojectTomlConfigSettingsSource,
  32. SecretsSettingsSource,
  33. TomlConfigSettingsSource,
  34. YamlConfigSettingsSource,
  35. get_subcommand,
  36. )
  37. from .sources.utils import _get_alias_names
  38. T = TypeVar('T')
  39. class SettingsConfigDict(ConfigDict, total=False):
  40. case_sensitive: bool
  41. nested_model_default_partial_update: bool | None
  42. env_prefix: str
  43. env_prefix_target: EnvPrefixTarget
  44. env_file: DotenvType | None
  45. env_file_encoding: str | None
  46. env_ignore_empty: bool
  47. env_nested_delimiter: str | None
  48. env_nested_max_split: int | None
  49. env_parse_none_str: str | None
  50. env_parse_enums: bool | None
  51. cli_prog_name: str | None
  52. cli_parse_args: bool | list[str] | tuple[str, ...] | None
  53. cli_parse_none_str: str | None
  54. cli_hide_none_type: bool
  55. cli_avoid_json: bool
  56. cli_enforce_required: bool
  57. cli_use_class_docs_for_groups: bool
  58. cli_exit_on_error: bool
  59. cli_prefix: str
  60. cli_flag_prefix_char: str
  61. cli_implicit_flags: bool | Literal['dual', 'toggle'] | None
  62. cli_ignore_unknown_args: bool | None
  63. cli_kebab_case: bool | Literal['all', 'no_enums'] | None
  64. cli_shortcuts: Mapping[str, str | list[str]] | None
  65. secrets_dir: PathType | None
  66. json_file: PathType | None
  67. json_file_encoding: str | None
  68. yaml_file: PathType | None
  69. yaml_file_encoding: str | None
  70. yaml_config_section: str | None
  71. """
  72. Specifies the section in a YAML file from which to load the settings.
  73. Supports dot-notation for nested paths (e.g., 'config.app.settings').
  74. If provided, the settings will be loaded from the specified section.
  75. This is useful when the YAML file contains multiple configuration sections
  76. and you only want to load a specific subset into your settings model.
  77. """
  78. pyproject_toml_depth: int
  79. """
  80. Number of levels **up** from the current working directory to attempt to find a pyproject.toml
  81. file.
  82. This is only used when a pyproject.toml file is not found in the current working directory.
  83. """
  84. pyproject_toml_table_header: tuple[str, ...]
  85. """
  86. Header of the TOML table within a pyproject.toml file to use when filling variables.
  87. This is supplied as a `tuple[str, ...]` instead of a `str` to accommodate for headers
  88. containing a `.`.
  89. For example, `toml_table_header = ("tool", "my.tool", "foo")` can be used to fill variable
  90. values from a table with header `[tool."my.tool".foo]`.
  91. To use the root table, exclude this config setting or provide an empty tuple.
  92. """
  93. toml_file: PathType | None
  94. enable_decoding: bool
  95. # Extend `config_keys` by pydantic settings config keys to
  96. # support setting config through class kwargs.
  97. # Pydantic uses `config_keys` in `pydantic._internal._config.ConfigWrapper.for_model`
  98. # to extract config keys from model kwargs, So, by adding pydantic settings keys to
  99. # `config_keys`, they will be considered as valid config keys and will be collected
  100. # by Pydantic.
  101. config_keys |= set(SettingsConfigDict.__annotations__.keys())
  102. class BaseSettings(BaseModel):
  103. """
  104. Base class for settings, allowing values to be overridden by environment variables.
  105. This is useful in production for secrets you do not wish to save in code, it plays nicely with docker(-compose),
  106. Heroku and any 12 factor app design.
  107. All the below attributes can be set via `model_config`.
  108. Args:
  109. _case_sensitive: Whether environment and CLI variable names should be read with case-sensitivity.
  110. Defaults to `None`.
  111. _nested_model_default_partial_update: Whether to allow partial updates on nested model default object fields.
  112. Defaults to `False`.
  113. _env_prefix: Prefix for all environment variables. Defaults to `None`.
  114. _env_prefix_target: Targets to which `_env_prefix` is applied. Default: `variable`.
  115. _env_file: The env file(s) to load settings values from. Defaults to `Path('')`, which
  116. means that the value from `model_config['env_file']` should be used. You can also pass
  117. `None` to indicate that environment variables should not be loaded from an env file.
  118. _env_file_encoding: The env file encoding, e.g. `'latin-1'`. Defaults to `None`.
  119. _env_ignore_empty: Ignore environment variables where the value is an empty string. Default to `False`.
  120. _env_nested_delimiter: The nested env values delimiter. Defaults to `None`.
  121. _env_nested_max_split: The nested env values maximum nesting. Defaults to `None`, which means no limit.
  122. _env_parse_none_str: The env string value that should be parsed (e.g. "null", "void", "None", etc.)
  123. into `None` type(None). Defaults to `None` type(None), which means no parsing should occur.
  124. _env_parse_enums: Parse enum field names to values. Defaults to `None.`, which means no parsing should occur.
  125. _cli_prog_name: The CLI program name to display in help text. Defaults to `None` if _cli_parse_args is `None`.
  126. Otherwise, defaults to sys.argv[0].
  127. _cli_parse_args: The list of CLI arguments to parse. Defaults to None.
  128. If set to `True`, defaults to sys.argv[1:].
  129. _cli_settings_source: Override the default CLI settings source with a user defined instance. Defaults to None.
  130. _cli_parse_none_str: The CLI string value that should be parsed (e.g. "null", "void", "None", etc.) into
  131. `None` type(None). Defaults to _env_parse_none_str value if set. Otherwise, defaults to "null" if
  132. _cli_avoid_json is `False`, and "None" if _cli_avoid_json is `True`.
  133. _cli_hide_none_type: Hide `None` values in CLI help text. Defaults to `False`.
  134. _cli_avoid_json: Avoid complex JSON objects in CLI help text. Defaults to `False`.
  135. _cli_enforce_required: Enforce required fields at the CLI. Defaults to `False`.
  136. _cli_use_class_docs_for_groups: Use class docstrings in CLI group help text instead of field descriptions.
  137. Defaults to `False`.
  138. _cli_exit_on_error: Determines whether or not the internal parser exits with error info when an error occurs.
  139. Defaults to `True`.
  140. _cli_prefix: The root parser command line arguments prefix. Defaults to "".
  141. _cli_flag_prefix_char: The flag prefix character to use for CLI optional arguments. Defaults to '-'.
  142. _cli_implicit_flags: Controls how `bool` fields are exposed as CLI flags.
  143. - False (default): no implicit flags are generated; booleans must be set explicitly (e.g. --flag=true).
  144. - True / 'dual': optional boolean fields generate both positive and negative forms (--flag and --no-flag).
  145. - 'toggle': required boolean fields remain in 'dual' mode, while optional boolean fields generate a single
  146. flag aligned with the default value (if default=False, expose --flag; if default=True, expose --no-flag).
  147. _cli_ignore_unknown_args: Whether to ignore unknown CLI args and parse only known ones. Defaults to `False`.
  148. _cli_kebab_case: CLI args use kebab case. Defaults to `False`.
  149. _cli_shortcuts: Mapping of target field name to alias names. Defaults to `None`.
  150. _secrets_dir: The secret files directory or a sequence of directories. Defaults to `None`.
  151. _build_sources: Pre-initialized sources and init kwargs to use for building instantiation values.
  152. Defaults to `None`.
  153. """
  154. def __init__(
  155. __pydantic_self__,
  156. _case_sensitive: bool | None = None,
  157. _nested_model_default_partial_update: bool | None = None,
  158. _env_prefix: str | None = None,
  159. _env_prefix_target: EnvPrefixTarget | None = None,
  160. _env_file: DotenvType | None = ENV_FILE_SENTINEL,
  161. _env_file_encoding: str | None = None,
  162. _env_ignore_empty: bool | None = None,
  163. _env_nested_delimiter: str | None = None,
  164. _env_nested_max_split: int | None = None,
  165. _env_parse_none_str: str | None = None,
  166. _env_parse_enums: bool | None = None,
  167. _cli_prog_name: str | None = None,
  168. _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None,
  169. _cli_settings_source: CliSettingsSource[Any] | None = None,
  170. _cli_parse_none_str: str | None = None,
  171. _cli_hide_none_type: bool | None = None,
  172. _cli_avoid_json: bool | None = None,
  173. _cli_enforce_required: bool | None = None,
  174. _cli_use_class_docs_for_groups: bool | None = None,
  175. _cli_exit_on_error: bool | None = None,
  176. _cli_prefix: str | None = None,
  177. _cli_flag_prefix_char: str | None = None,
  178. _cli_implicit_flags: bool | Literal['dual', 'toggle'] | None = None,
  179. _cli_ignore_unknown_args: bool | None = None,
  180. _cli_kebab_case: bool | Literal['all', 'no_enums'] | None = None,
  181. _cli_shortcuts: Mapping[str, str | list[str]] | None = None,
  182. _secrets_dir: PathType | None = None,
  183. _build_sources: tuple[tuple[PydanticBaseSettingsSource, ...], dict[str, Any]] | None = None,
  184. **values: Any,
  185. ) -> None:
  186. sources, init_kwargs = (
  187. _build_sources
  188. if _build_sources is not None
  189. else __pydantic_self__.__class__._settings_init_sources(
  190. _case_sensitive=_case_sensitive,
  191. _nested_model_default_partial_update=_nested_model_default_partial_update,
  192. _env_prefix=_env_prefix,
  193. _env_prefix_target=_env_prefix_target,
  194. _env_file=_env_file,
  195. _env_file_encoding=_env_file_encoding,
  196. _env_ignore_empty=_env_ignore_empty,
  197. _env_nested_delimiter=_env_nested_delimiter,
  198. _env_nested_max_split=_env_nested_max_split,
  199. _env_parse_none_str=_env_parse_none_str,
  200. _env_parse_enums=_env_parse_enums,
  201. _cli_prog_name=_cli_prog_name,
  202. _cli_parse_args=_cli_parse_args,
  203. _cli_settings_source=_cli_settings_source,
  204. _cli_parse_none_str=_cli_parse_none_str,
  205. _cli_hide_none_type=_cli_hide_none_type,
  206. _cli_avoid_json=_cli_avoid_json,
  207. _cli_enforce_required=_cli_enforce_required,
  208. _cli_use_class_docs_for_groups=_cli_use_class_docs_for_groups,
  209. _cli_exit_on_error=_cli_exit_on_error,
  210. _cli_prefix=_cli_prefix,
  211. _cli_flag_prefix_char=_cli_flag_prefix_char,
  212. _cli_implicit_flags=_cli_implicit_flags,
  213. _cli_ignore_unknown_args=_cli_ignore_unknown_args,
  214. _cli_kebab_case=_cli_kebab_case,
  215. _cli_shortcuts=_cli_shortcuts,
  216. _secrets_dir=_secrets_dir,
  217. **values,
  218. )
  219. )
  220. super().__init__(**__pydantic_self__.__class__._settings_build_values(sources, init_kwargs))
  221. @classmethod
  222. def settings_customise_sources(
  223. cls,
  224. settings_cls: type[BaseSettings],
  225. init_settings: PydanticBaseSettingsSource,
  226. env_settings: PydanticBaseSettingsSource,
  227. dotenv_settings: PydanticBaseSettingsSource,
  228. file_secret_settings: PydanticBaseSettingsSource,
  229. ) -> tuple[PydanticBaseSettingsSource, ...]:
  230. """
  231. Define the sources and their order for loading the settings values.
  232. Args:
  233. settings_cls: The Settings class.
  234. init_settings: The `InitSettingsSource` instance.
  235. env_settings: The `EnvSettingsSource` instance.
  236. dotenv_settings: The `DotEnvSettingsSource` instance.
  237. file_secret_settings: The `SecretsSettingsSource` instance.
  238. Returns:
  239. A tuple containing the sources and their order for loading the settings values.
  240. """
  241. return init_settings, env_settings, dotenv_settings, file_secret_settings
  242. @classmethod
  243. def _settings_init_sources(
  244. cls,
  245. _case_sensitive: bool | None = None,
  246. _nested_model_default_partial_update: bool | None = None,
  247. _env_prefix: str | None = None,
  248. _env_prefix_target: EnvPrefixTarget | None = None,
  249. _env_file: DotenvType | None = None,
  250. _env_file_encoding: str | None = None,
  251. _env_ignore_empty: bool | None = None,
  252. _env_nested_delimiter: str | None = None,
  253. _env_nested_max_split: int | None = None,
  254. _env_parse_none_str: str | None = None,
  255. _env_parse_enums: bool | None = None,
  256. _cli_prog_name: str | None = None,
  257. _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None,
  258. _cli_settings_source: CliSettingsSource[Any] | None = None,
  259. _cli_parse_none_str: str | None = None,
  260. _cli_hide_none_type: bool | None = None,
  261. _cli_avoid_json: bool | None = None,
  262. _cli_enforce_required: bool | None = None,
  263. _cli_use_class_docs_for_groups: bool | None = None,
  264. _cli_exit_on_error: bool | None = None,
  265. _cli_prefix: str | None = None,
  266. _cli_flag_prefix_char: str | None = None,
  267. _cli_implicit_flags: bool | Literal['dual', 'toggle'] | None = None,
  268. _cli_ignore_unknown_args: bool | None = None,
  269. _cli_kebab_case: bool | Literal['all', 'no_enums'] | None = None,
  270. _cli_shortcuts: Mapping[str, str | list[str]] | None = None,
  271. _secrets_dir: PathType | None = None,
  272. **init_kwargs: dict[str, Any],
  273. ) -> tuple[tuple[PydanticBaseSettingsSource, ...], dict[str, Any]]:
  274. # Determine settings config values
  275. case_sensitive = _case_sensitive if _case_sensitive is not None else cls.model_config.get('case_sensitive')
  276. env_prefix = _env_prefix if _env_prefix is not None else cls.model_config.get('env_prefix')
  277. env_prefix_target = (
  278. _env_prefix_target if _env_prefix_target is not None else cls.model_config.get('env_prefix_target')
  279. )
  280. nested_model_default_partial_update = (
  281. _nested_model_default_partial_update
  282. if _nested_model_default_partial_update is not None
  283. else cls.model_config.get('nested_model_default_partial_update')
  284. )
  285. env_file = _env_file if _env_file != ENV_FILE_SENTINEL else cls.model_config.get('env_file')
  286. env_file_encoding = (
  287. _env_file_encoding if _env_file_encoding is not None else cls.model_config.get('env_file_encoding')
  288. )
  289. env_ignore_empty = (
  290. _env_ignore_empty if _env_ignore_empty is not None else cls.model_config.get('env_ignore_empty')
  291. )
  292. env_nested_delimiter = (
  293. _env_nested_delimiter if _env_nested_delimiter is not None else cls.model_config.get('env_nested_delimiter')
  294. )
  295. env_nested_max_split = (
  296. _env_nested_max_split if _env_nested_max_split is not None else cls.model_config.get('env_nested_max_split')
  297. )
  298. env_parse_none_str = (
  299. _env_parse_none_str if _env_parse_none_str is not None else cls.model_config.get('env_parse_none_str')
  300. )
  301. env_parse_enums = _env_parse_enums if _env_parse_enums is not None else cls.model_config.get('env_parse_enums')
  302. cli_prog_name = _cli_prog_name if _cli_prog_name is not None else cls.model_config.get('cli_prog_name')
  303. cli_parse_args = _cli_parse_args if _cli_parse_args is not None else cls.model_config.get('cli_parse_args')
  304. cli_settings_source = (
  305. _cli_settings_source if _cli_settings_source is not None else cls.model_config.get('cli_settings_source')
  306. )
  307. cli_parse_none_str = (
  308. _cli_parse_none_str if _cli_parse_none_str is not None else cls.model_config.get('cli_parse_none_str')
  309. )
  310. cli_parse_none_str = cli_parse_none_str if not env_parse_none_str else env_parse_none_str
  311. cli_hide_none_type = (
  312. _cli_hide_none_type if _cli_hide_none_type is not None else cls.model_config.get('cli_hide_none_type')
  313. )
  314. cli_avoid_json = _cli_avoid_json if _cli_avoid_json is not None else cls.model_config.get('cli_avoid_json')
  315. cli_enforce_required = (
  316. _cli_enforce_required if _cli_enforce_required is not None else cls.model_config.get('cli_enforce_required')
  317. )
  318. cli_use_class_docs_for_groups = (
  319. _cli_use_class_docs_for_groups
  320. if _cli_use_class_docs_for_groups is not None
  321. else cls.model_config.get('cli_use_class_docs_for_groups')
  322. )
  323. cli_exit_on_error = (
  324. _cli_exit_on_error if _cli_exit_on_error is not None else cls.model_config.get('cli_exit_on_error')
  325. )
  326. cli_prefix = _cli_prefix if _cli_prefix is not None else cls.model_config.get('cli_prefix')
  327. cli_flag_prefix_char = (
  328. _cli_flag_prefix_char if _cli_flag_prefix_char is not None else cls.model_config.get('cli_flag_prefix_char')
  329. )
  330. cli_implicit_flags = (
  331. _cli_implicit_flags if _cli_implicit_flags is not None else cls.model_config.get('cli_implicit_flags')
  332. )
  333. cli_ignore_unknown_args = (
  334. _cli_ignore_unknown_args
  335. if _cli_ignore_unknown_args is not None
  336. else cls.model_config.get('cli_ignore_unknown_args')
  337. )
  338. cli_kebab_case = _cli_kebab_case if _cli_kebab_case is not None else cls.model_config.get('cli_kebab_case')
  339. cli_shortcuts = _cli_shortcuts if _cli_shortcuts is not None else cls.model_config.get('cli_shortcuts')
  340. secrets_dir = _secrets_dir if _secrets_dir is not None else cls.model_config.get('secrets_dir')
  341. # Configure built-in sources
  342. default_settings = DefaultSettingsSource(
  343. cls, nested_model_default_partial_update=nested_model_default_partial_update
  344. )
  345. init_settings = InitSettingsSource(
  346. cls,
  347. init_kwargs=init_kwargs,
  348. nested_model_default_partial_update=nested_model_default_partial_update,
  349. )
  350. env_settings = EnvSettingsSource(
  351. cls,
  352. case_sensitive=case_sensitive,
  353. env_prefix=env_prefix,
  354. env_prefix_target=env_prefix_target,
  355. env_nested_delimiter=env_nested_delimiter,
  356. env_nested_max_split=env_nested_max_split,
  357. env_ignore_empty=env_ignore_empty,
  358. env_parse_none_str=env_parse_none_str,
  359. env_parse_enums=env_parse_enums,
  360. )
  361. dotenv_settings = DotEnvSettingsSource(
  362. cls,
  363. env_file=env_file,
  364. env_file_encoding=env_file_encoding,
  365. case_sensitive=case_sensitive,
  366. env_prefix=env_prefix,
  367. env_prefix_target=env_prefix_target,
  368. env_nested_delimiter=env_nested_delimiter,
  369. env_nested_max_split=env_nested_max_split,
  370. env_ignore_empty=env_ignore_empty,
  371. env_parse_none_str=env_parse_none_str,
  372. env_parse_enums=env_parse_enums,
  373. )
  374. file_secret_settings = SecretsSettingsSource(
  375. cls,
  376. secrets_dir=secrets_dir,
  377. case_sensitive=case_sensitive,
  378. env_prefix=env_prefix,
  379. env_prefix_target=env_prefix_target,
  380. )
  381. # Provide a hook to set built-in sources priority and add / remove sources
  382. sources = cls.settings_customise_sources(
  383. cls,
  384. init_settings=init_settings,
  385. env_settings=env_settings,
  386. dotenv_settings=dotenv_settings,
  387. file_secret_settings=file_secret_settings,
  388. ) + (default_settings,)
  389. custom_cli_sources = [source for source in sources if isinstance(source, CliSettingsSource)]
  390. if not any(custom_cli_sources):
  391. if isinstance(cli_settings_source, CliSettingsSource):
  392. sources = (cli_settings_source,) + sources
  393. elif cli_parse_args is not None:
  394. cli_settings = CliSettingsSource[Any](
  395. cls,
  396. cli_prog_name=cli_prog_name,
  397. cli_parse_args=cli_parse_args,
  398. cli_parse_none_str=cli_parse_none_str,
  399. cli_hide_none_type=cli_hide_none_type,
  400. cli_avoid_json=cli_avoid_json,
  401. cli_enforce_required=cli_enforce_required,
  402. cli_use_class_docs_for_groups=cli_use_class_docs_for_groups,
  403. cli_exit_on_error=cli_exit_on_error,
  404. cli_prefix=cli_prefix,
  405. cli_flag_prefix_char=cli_flag_prefix_char,
  406. cli_implicit_flags=cli_implicit_flags,
  407. cli_ignore_unknown_args=cli_ignore_unknown_args,
  408. cli_kebab_case=cli_kebab_case,
  409. cli_shortcuts=cli_shortcuts,
  410. case_sensitive=case_sensitive,
  411. )
  412. sources = (cli_settings,) + sources
  413. # We ensure that if command line arguments haven't been parsed yet, we do so.
  414. elif cli_parse_args not in (None, False) and not custom_cli_sources[0].env_vars:
  415. custom_cli_sources[0](args=cli_parse_args) # type: ignore
  416. cls._settings_warn_unused_config_keys(sources, cls.model_config)
  417. return sources, init_kwargs
  418. @classmethod
  419. def _settings_build_values(
  420. cls, sources: tuple[PydanticBaseSettingsSource, ...], init_kwargs: dict[str, Any]
  421. ) -> dict[str, Any]:
  422. if sources:
  423. state: dict[str, Any] = {}
  424. defaults: dict[str, Any] = {}
  425. states: dict[str, dict[str, Any]] = {}
  426. for source in sources:
  427. if isinstance(source, PydanticBaseSettingsSource):
  428. source._set_current_state(state)
  429. source._set_settings_sources_data(states)
  430. source_name = source.__name__ if hasattr(source, '__name__') else type(source).__name__
  431. source_state = source()
  432. if isinstance(source, DefaultSettingsSource):
  433. defaults = source_state
  434. states[source_name] = source_state
  435. state = deep_update(source_state, state)
  436. # Strip any default values not explicity set before returning final state
  437. state = {key: val for key, val in state.items() if key not in defaults or defaults[key] != val}
  438. cls._settings_restore_init_kwarg_names(cls, init_kwargs, state)
  439. return state
  440. else:
  441. # no one should mean to do this, but I think returning an empty dict is marginally preferable
  442. # to an informative error and much better than a confusing error
  443. return {}
  444. @staticmethod
  445. def _settings_restore_init_kwarg_names(
  446. settings_cls: type[BaseSettings], init_kwargs: dict[str, Any], state: dict[str, Any]
  447. ) -> None:
  448. """
  449. Restore the init_kwarg key names to the final merged state dictionary.
  450. This function renames keys in state to match the original init_kwargs key names,
  451. preserving the merged values from the source priority order.
  452. """
  453. if init_kwargs and state:
  454. state_kwarg_names = set(state.keys())
  455. init_kwarg_names = set(init_kwargs.keys())
  456. for field_name, field_info in settings_cls.model_fields.items():
  457. alias_names, *_ = _get_alias_names(field_name, field_info)
  458. matchable_names = set(alias_names)
  459. include_name = settings_cls.model_config.get(
  460. 'populate_by_name', False
  461. ) or settings_cls.model_config.get('validate_by_name', False)
  462. if include_name:
  463. matchable_names.add(field_name)
  464. init_kwarg_name = init_kwarg_names & matchable_names
  465. state_kwarg_name = state_kwarg_names & matchable_names
  466. if init_kwarg_name and state_kwarg_name:
  467. # Use deterministic selection for both keys.
  468. # Target key: the key from init_kwargs that should be used in the final state.
  469. target_key = next(iter(init_kwarg_name))
  470. # Source key: prefer the alias (first in alias_names) if present in state,
  471. # as InitSettingsSource normalizes to the preferred alias.
  472. # This ensures we get the highest-priority value for this field.
  473. source_key = None
  474. for alias in alias_names:
  475. if alias in state_kwarg_name:
  476. source_key = alias
  477. break
  478. if source_key is None:
  479. # Fall back to field_name if no alias found in state
  480. source_key = field_name if field_name in state_kwarg_name else next(iter(state_kwarg_name))
  481. # Get the value from the source key and remove all matching keys
  482. value = state.pop(source_key)
  483. for key in state_kwarg_name - {source_key}:
  484. state.pop(key, None)
  485. state[target_key] = value
  486. @staticmethod
  487. def _settings_warn_unused_config_keys(sources: tuple[object, ...], model_config: SettingsConfigDict) -> None:
  488. """
  489. Warns if any values in model_config were set but the corresponding settings source has not been initialised.
  490. The list alternative sources and their config keys can be found here:
  491. https://docs.pydantic.dev/latest/concepts/pydantic_settings/#other-settings-source
  492. Args:
  493. sources: The tuple of configured sources
  494. model_config: The model config to check for unused config keys
  495. """
  496. def warn_if_not_used(source_type: type[PydanticBaseSettingsSource], keys: tuple[str, ...]) -> None:
  497. if not any(isinstance(source, source_type) for source in sources):
  498. for key in keys:
  499. if model_config.get(key) is not None:
  500. warnings.warn(
  501. f'Config key `{key}` is set in model_config but will be ignored because no '
  502. f'{source_type.__name__} source is configured. To use this config key, add a '
  503. f'{source_type.__name__} source to the settings sources via the '
  504. 'settings_customise_sources hook.',
  505. UserWarning,
  506. stacklevel=3,
  507. )
  508. warn_if_not_used(JsonConfigSettingsSource, ('json_file', 'json_file_encoding'))
  509. warn_if_not_used(PyprojectTomlConfigSettingsSource, ('pyproject_toml_depth', 'pyproject_toml_table_header'))
  510. warn_if_not_used(TomlConfigSettingsSource, ('toml_file',))
  511. warn_if_not_used(YamlConfigSettingsSource, ('yaml_file', 'yaml_file_encoding', 'yaml_config_section'))
  512. model_config: ClassVar[SettingsConfigDict] = SettingsConfigDict(
  513. extra='forbid',
  514. arbitrary_types_allowed=True,
  515. validate_default=True,
  516. case_sensitive=False,
  517. env_prefix='',
  518. env_prefix_target='variable',
  519. nested_model_default_partial_update=False,
  520. env_file=None,
  521. env_file_encoding=None,
  522. env_ignore_empty=False,
  523. env_nested_delimiter=None,
  524. env_nested_max_split=None,
  525. env_parse_none_str=None,
  526. env_parse_enums=None,
  527. cli_prog_name=None,
  528. cli_parse_args=None,
  529. cli_parse_none_str=None,
  530. cli_hide_none_type=False,
  531. cli_avoid_json=False,
  532. cli_enforce_required=False,
  533. cli_use_class_docs_for_groups=False,
  534. cli_exit_on_error=True,
  535. cli_prefix='',
  536. cli_flag_prefix_char='-',
  537. cli_implicit_flags=False,
  538. cli_ignore_unknown_args=False,
  539. cli_kebab_case=False,
  540. cli_shortcuts=None,
  541. json_file=None,
  542. json_file_encoding=None,
  543. yaml_file=None,
  544. yaml_file_encoding=None,
  545. yaml_config_section=None,
  546. toml_file=None,
  547. secrets_dir=None,
  548. protected_namespaces=('model_validate', 'model_dump', 'settings_customise_sources'),
  549. enable_decoding=True,
  550. )
  551. class CliApp:
  552. """
  553. A utility class for running Pydantic `BaseSettings`, `BaseModel`, or `pydantic.dataclasses.dataclass` as
  554. CLI applications.
  555. """
  556. _subcommand_stack: ClassVar[dict[int, tuple[CliSettingsSource[Any], Any, str]]] = {}
  557. _ansi_color: ClassVar[re.Pattern[str]] = re.compile(r'\x1b\[[0-9;]*m')
  558. @staticmethod
  559. def _get_base_settings_cls(model_cls: type[Any]) -> type[BaseSettings]:
  560. if issubclass(model_cls, BaseSettings):
  561. return model_cls
  562. class CliAppBaseSettings(BaseSettings, model_cls): # type: ignore
  563. __doc__ = model_cls.__doc__
  564. model_config = SettingsConfigDict(
  565. nested_model_default_partial_update=True,
  566. case_sensitive=True,
  567. cli_hide_none_type=True,
  568. cli_avoid_json=True,
  569. cli_enforce_required=True,
  570. cli_implicit_flags=True,
  571. cli_kebab_case=True,
  572. )
  573. return CliAppBaseSettings
  574. @staticmethod
  575. def _run_cli_cmd(model: Any, cli_cmd_method_name: str, is_required: bool) -> Any:
  576. command = getattr(type(model), cli_cmd_method_name, None)
  577. if command is None:
  578. if is_required:
  579. raise SettingsError(f'Error: {type(model).__name__} class is missing {cli_cmd_method_name} entrypoint')
  580. return model
  581. # If the method is asynchronous, we handle its execution based on the current event loop status.
  582. if inspect.iscoroutinefunction(command):
  583. # For asynchronous methods, we have two execution scenarios:
  584. # 1. If no event loop is running in the current thread, run the coroutine directly with asyncio.run().
  585. # 2. If an event loop is already running in the current thread, run the coroutine in a separate thread to avoid conflicts.
  586. try:
  587. # Check if an event loop is currently running in this thread.
  588. loop = asyncio.get_running_loop()
  589. except RuntimeError:
  590. loop = None
  591. if loop and loop.is_running():
  592. # We're in a context with an active event loop (e.g., Jupyter Notebook).
  593. # Running asyncio.run() here would cause conflicts, so we use a separate thread.
  594. exception_container = []
  595. def run_coro() -> None:
  596. try:
  597. # Execute the coroutine in a new event loop in this separate thread.
  598. asyncio.run(command(model))
  599. except Exception as e:
  600. exception_container.append(e)
  601. thread = threading.Thread(target=run_coro)
  602. thread.start()
  603. thread.join()
  604. if exception_container:
  605. # Propagate exceptions from the separate thread.
  606. raise exception_container[0]
  607. else:
  608. # No event loop is running; safe to run the coroutine directly.
  609. asyncio.run(command(model))
  610. else:
  611. # For synchronous methods, call them directly.
  612. command(model)
  613. return model
  614. @staticmethod
  615. def run(
  616. model_cls: type[T],
  617. cli_args: list[str] | Namespace | SimpleNamespace | dict[str, Any] | None = None,
  618. cli_settings_source: CliSettingsSource[Any] | None = None,
  619. cli_exit_on_error: bool | None = None,
  620. cli_cmd_method_name: str = 'cli_cmd',
  621. **model_init_data: Any,
  622. ) -> T:
  623. """
  624. Runs a Pydantic `BaseSettings`, `BaseModel`, or `pydantic.dataclasses.dataclass` as a CLI application.
  625. Running a model as a CLI application requires the `cli_cmd` method to be defined in the model class.
  626. Args:
  627. model_cls: The model class to run as a CLI application.
  628. cli_args: The list of CLI arguments to parse. If `cli_settings_source` is specified, this may
  629. also be a namespace or dictionary of pre-parsed CLI arguments. Defaults to `sys.argv[1:]`.
  630. cli_settings_source: Override the default CLI settings source with a user defined instance.
  631. Defaults to `None`.
  632. cli_exit_on_error: Determines whether this function exits on error. If model is subclass of
  633. `BaseSettings`, defaults to BaseSettings `cli_exit_on_error` value. Otherwise, defaults to
  634. `True`.
  635. cli_cmd_method_name: The CLI command method name to run. Defaults to "cli_cmd".
  636. model_init_data: The model init data.
  637. Returns:
  638. The ran instance of model.
  639. Raises:
  640. SettingsError: If model_cls is not subclass of `BaseModel` or `pydantic.dataclasses.dataclass`.
  641. SettingsError: If model_cls does not have a `cli_cmd` entrypoint defined.
  642. """
  643. if not (is_pydantic_dataclass(model_cls) or is_model_class(model_cls)):
  644. raise SettingsError(
  645. f'Error: {model_cls.__name__} is not subclass of BaseModel or pydantic.dataclasses.dataclass'
  646. )
  647. cli_settings = None
  648. cli_parse_args = True if cli_args is None else cli_args
  649. if cli_settings_source is not None:
  650. if isinstance(cli_parse_args, (Namespace, SimpleNamespace, dict)):
  651. cli_settings = cli_settings_source(parsed_args=cli_parse_args)
  652. else:
  653. cli_settings = cli_settings_source(args=cli_parse_args)
  654. elif isinstance(cli_parse_args, (Namespace, SimpleNamespace, dict)):
  655. raise SettingsError('Error: `cli_args` must be list[str] or None when `cli_settings_source` is not used')
  656. model_init_data['_cli_parse_args'] = cli_parse_args
  657. model_init_data['_cli_exit_on_error'] = cli_exit_on_error
  658. model_init_data['_cli_settings_source'] = cli_settings
  659. if not issubclass(model_cls, BaseSettings):
  660. base_settings_cls = CliApp._get_base_settings_cls(model_cls)
  661. sources, init_kwargs = base_settings_cls._settings_init_sources(**model_init_data)
  662. model = base_settings_cls(**base_settings_cls._settings_build_values(sources, init_kwargs))
  663. model_init_data = {}
  664. for field_name, field_info in base_settings_cls.model_fields.items():
  665. model_init_data[_field_name_for_signature(field_name, field_info)] = getattr(model, field_name)
  666. command = model_cls(**model_init_data)
  667. else:
  668. sources, init_kwargs = model_cls._settings_init_sources(**model_init_data)
  669. command = model_cls(_build_sources=(sources, init_kwargs))
  670. subcommand_dest = ':subcommand'
  671. cli_settings_source = [source for source in sources if isinstance(source, CliSettingsSource)][0]
  672. CliApp._subcommand_stack[id(command)] = (cli_settings_source, cli_settings_source.root_parser, subcommand_dest)
  673. try:
  674. data_model = CliApp._run_cli_cmd(command, cli_cmd_method_name, is_required=False)
  675. finally:
  676. del CliApp._subcommand_stack[id(command)]
  677. return data_model
  678. @staticmethod
  679. def run_subcommand(
  680. model: PydanticModel, cli_exit_on_error: bool | None = None, cli_cmd_method_name: str = 'cli_cmd'
  681. ) -> PydanticModel:
  682. """
  683. Runs the model subcommand. Running a model subcommand requires the `cli_cmd` method to be defined in
  684. the nested model subcommand class.
  685. Args:
  686. model: The model to run the subcommand from.
  687. cli_exit_on_error: Determines whether this function exits with error if no subcommand is found.
  688. Defaults to model_config `cli_exit_on_error` value if set. Otherwise, defaults to `True`.
  689. cli_cmd_method_name: The CLI command method name to run. Defaults to "cli_cmd".
  690. Returns:
  691. The ran subcommand model.
  692. Raises:
  693. SystemExit: When no subcommand is found and cli_exit_on_error=`True` (the default).
  694. SettingsError: When no subcommand is found and cli_exit_on_error=`False`.
  695. """
  696. if id(model) in CliApp._subcommand_stack:
  697. cli_settings_source, parser, subcommand_dest = CliApp._subcommand_stack[id(model)]
  698. else:
  699. cli_settings_source = CliSettingsSource[Any](CliApp._get_base_settings_cls(type(model)))
  700. parser = cli_settings_source.root_parser
  701. subcommand_dest = ':subcommand'
  702. cli_exit_on_error = cli_settings_source.cli_exit_on_error if cli_exit_on_error is None else cli_exit_on_error
  703. errors: list[SettingsError | SystemExit] = []
  704. subcommand = get_subcommand(
  705. model, is_required=True, cli_exit_on_error=cli_exit_on_error, _suppress_errors=errors
  706. )
  707. if errors:
  708. err = errors[0]
  709. if err.__context__ is None and err.__cause__ is None and cli_settings_source._format_help is not None:
  710. error_message = f'{err}\n{cli_settings_source._format_help(parser)}'
  711. raise type(err)(error_message) from None
  712. else:
  713. raise err
  714. subcommand_cls = cast(type[BaseModel], type(subcommand))
  715. subcommand_arg = cli_settings_source._parser_map[subcommand_dest][subcommand_cls]
  716. subcommand_alias = subcommand_arg.subcommand_alias(subcommand_cls)
  717. subcommand_dest = f'{subcommand_dest.split(":")[0]}{subcommand_alias}.:subcommand'
  718. subcommand_parser = subcommand_arg.parser
  719. CliApp._subcommand_stack[id(subcommand)] = (cli_settings_source, subcommand_parser, subcommand_dest)
  720. try:
  721. data_model = CliApp._run_cli_cmd(subcommand, cli_cmd_method_name, is_required=True)
  722. finally:
  723. del CliApp._subcommand_stack[id(subcommand)]
  724. return data_model
  725. @staticmethod
  726. def serialize(
  727. model: PydanticModel,
  728. list_style: Literal['json', 'argparse', 'lazy'] = 'json',
  729. dict_style: Literal['json', 'env'] = 'json',
  730. positionals_first: bool = False,
  731. ) -> list[str]:
  732. """
  733. Serializes the CLI arguments for a Pydantic data model.
  734. Args:
  735. model: The data model to serialize.
  736. list_style:
  737. Controls how list-valued fields are serialized on the command line.
  738. - 'json' (default):
  739. Lists are encoded as a single JSON array.
  740. Example: `--tags '["a","b","c"]'`
  741. - 'argparse':
  742. Each list element becomes its own repeated flag, following
  743. typical `argparse` conventions.
  744. Example: `--tags a --tags b --tags c`
  745. - 'lazy':
  746. Lists are emitted as a single comma-separated string without JSON
  747. quoting or escaping.
  748. Example: `--tags a,b,c`
  749. dict_style:
  750. Controls how dictionary-valued fields are serialized.
  751. - 'json' (default):
  752. The entire dictionary is emitted as a single JSON object.
  753. Example: `--config '{"host": "localhost", "port": 5432}'`
  754. - 'env':
  755. The dictionary is flattened into multiple CLI flags using
  756. environment-variable-style assignement.
  757. Example: `--config host=localhost --config port=5432`
  758. positionals_first: Controls whether positional arguments should be serialized
  759. first compared to optional arguments. Defaults to `False`.
  760. Returns:
  761. The serialized CLI arguments for the data model.
  762. """
  763. base_settings_cls = CliApp._get_base_settings_cls(type(model))
  764. serialized_args = CliSettingsSource[Any](base_settings_cls)._serialized_args(
  765. model,
  766. list_style=list_style,
  767. dict_style=dict_style,
  768. positionals_first=positionals_first,
  769. )
  770. return CliSettingsSource._flatten_serialized_args(serialized_args, positionals_first)
  771. @staticmethod
  772. def format_help(
  773. model: PydanticModel | type[T],
  774. cli_settings_source: CliSettingsSource[Any] | None = None,
  775. strip_ansi_color: bool = False,
  776. ) -> str:
  777. """
  778. Return a string containing a help message for a Pydantic model.
  779. Args:
  780. model: The model or model class.
  781. cli_settings_source: Override the default CLI settings source with a user defined instance.
  782. Defaults to `None`.
  783. strip_ansi_color: Strips ANSI color codes from the help message when set to `True`.
  784. Returns:
  785. The help message string for the model.
  786. """
  787. model_cls = model if isinstance(model, type) else type(model)
  788. if cli_settings_source is None:
  789. if not isinstance(model, type) and id(model) in CliApp._subcommand_stack:
  790. cli_settings_source, *_ = CliApp._subcommand_stack[id(model)]
  791. else:
  792. cli_settings_source = CliSettingsSource(CliApp._get_base_settings_cls(model_cls))
  793. help_message = cli_settings_source._format_help(cli_settings_source.root_parser)
  794. return help_message if not strip_ansi_color else CliApp._ansi_color.sub('', help_message)
  795. @staticmethod
  796. def print_help(
  797. model: PydanticModel | type[T],
  798. cli_settings_source: CliSettingsSource[Any] | None = None,
  799. file: TextIO | None = None,
  800. strip_ansi_color: bool = False,
  801. ) -> None:
  802. """
  803. Print a help message for a Pydantic model.
  804. Args:
  805. model: The model or model class.
  806. cli_settings_source: Override the default CLI settings source with a user defined instance.
  807. Defaults to `None`.
  808. file: A text stream to which the help message is written. If `None`, the output is sent to sys.stdout.
  809. strip_ansi_color: Strips ANSI color codes from the help message when set to `True`.
  810. """
  811. print(
  812. CliApp.format_help(
  813. model,
  814. cli_settings_source=cli_settings_source,
  815. strip_ansi_color=strip_ansi_color,
  816. ),
  817. file=file,
  818. )