codekingpro/portable-devtools
114k
1from __future__ import annotations as _annotations2 3import asyncio4import inspect5import re6import threading7import warnings8from argparse import Namespace9from collections.abc import Mapping10from types import SimpleNamespace11from typing import Any, ClassVar, Literal, TextIO, TypeVar, cast12 13from pydantic import ConfigDict14from pydantic._internal._config import config_keys15from pydantic._internal._signature import _field_name_for_signature16from pydantic._internal._utils import deep_update, is_model_class17from pydantic.dataclasses import is_pydantic_dataclass18from pydantic.main import BaseModel19 20from .exceptions import SettingsError21from .sources import (22 ENV_FILE_SENTINEL,23 CliSettingsSource,24 DefaultSettingsSource,25 DotenvFiltering,26 DotEnvSettingsSource,27 DotenvType,28 EnvPrefixTarget,29 EnvSettingsSource,30 InitSettingsSource,31 JsonConfigSettingsSource,32 PathType,33 PydanticBaseSettingsSource,34 PydanticModel,35 PyprojectTomlConfigSettingsSource,36 SecretsSettingsSource,37 TomlConfigSettingsSource,38 YamlConfigSettingsSource,39 get_subcommand,40)41from .sources.utils import _get_alias_names42 43T = TypeVar('T')44 45 46class SettingsConfigDict(ConfigDict, total=False):47 case_sensitive: bool48 nested_model_default_partial_update: bool | None49 env_prefix: str50 env_prefix_target: EnvPrefixTarget51 env_file: DotenvType | None52 env_file_encoding: str | None53 dotenv_filtering: DotenvFiltering | None54 env_ignore_empty: bool55 env_nested_delimiter: str | None56 env_nested_max_split: int | None57 env_parse_none_str: str | None58 env_parse_enums: bool | None59 cli_prog_name: str | None60 cli_parse_args: bool | list[str] | tuple[str, ...] | None61 cli_parse_none_str: str | None62 cli_hide_none_type: bool63 cli_avoid_json: bool64 cli_enforce_required: bool65 cli_use_class_docs_for_groups: bool66 cli_exit_on_error: bool67 cli_prefix: str68 cli_flag_prefix_char: str69 cli_implicit_flags: bool | Literal['dual', 'toggle'] | None70 cli_ignore_unknown_args: bool | None71 cli_kebab_case: bool | Literal['all', 'no_enums'] | None72 cli_shortcuts: Mapping[str, str | list[str]] | None73 secrets_dir: PathType | None74 json_file: PathType | None75 json_file_encoding: str | None76 yaml_file: PathType | None77 yaml_file_encoding: str | None78 yaml_config_section: str | None79 """80 Specifies the section in a YAML file from which to load the settings.81 Supports dot-notation for nested paths (e.g., 'config.app.settings').82 If provided, the settings will be loaded from the specified section.83 This is useful when the YAML file contains multiple configuration sections84 and you only want to load a specific subset into your settings model.85 """86 87 pyproject_toml_depth: int88 """89 Number of levels **up** from the current working directory to attempt to find a pyproject.toml90 file.91 92 This is only used when a pyproject.toml file is not found in the current working directory.93 """94 95 pyproject_toml_table_header: tuple[str, ...]96 """97 Header of the TOML table within a pyproject.toml file to use when filling variables.98 This is supplied as a `tuple[str, ...]` instead of a `str` to accommodate for headers99 containing a `.`.100 101 For example, `toml_table_header = ("tool", "my.tool", "foo")` can be used to fill variable102 values from a table with header `[tool."my.tool".foo]`.103 104 To use the root table, exclude this config setting or provide an empty tuple.105 """106 107 toml_file: PathType | None108 enable_decoding: bool109 110 111# Extend `config_keys` by pydantic settings config keys to112# support setting config through class kwargs.113# Pydantic uses `config_keys` in `pydantic._internal._config.ConfigWrapper.for_model`114# to extract config keys from model kwargs, So, by adding pydantic settings keys to115# `config_keys`, they will be considered as valid config keys and will be collected116# by Pydantic.117config_keys |= set(SettingsConfigDict.__annotations__.keys())118 119 120class BaseSettings(BaseModel):121 """122 Base class for settings, allowing values to be overridden by environment variables.123 124 This is useful in production for secrets you do not wish to save in code, it plays nicely with docker(-compose),125 Heroku and any 12 factor app design.126 127 All the below attributes can be set via `model_config`.128 129 Args:130 _case_sensitive: Whether environment and CLI variable names should be read with case-sensitivity.131 Defaults to `None`.132 _nested_model_default_partial_update: Whether to allow partial updates on nested model default object fields.133 Defaults to `False`.134 _env_prefix: Prefix for all environment variables. Defaults to `None`.135 _env_prefix_target: Targets to which `_env_prefix` is applied. Default: `variable`.136 _env_file: The env file(s) to load settings values from. Defaults to `Path('')`, which137 means that the value from `model_config['env_file']` should be used. You can also pass138 `None` to indicate that environment variables should not be loaded from an env file.139 _env_file_encoding: The env file encoding, e.g. `'latin-1'`. Defaults to `None`.140 _env_ignore_empty: Ignore environment variables where the value is an empty string. Default to `False`.141 _env_nested_delimiter: The nested env values delimiter. Defaults to `None`.142 _env_nested_max_split: The nested env values maximum nesting. Defaults to `None`, which means no limit.143 _env_parse_none_str: The env string value that should be parsed (e.g. "null", "void", "None", etc.)144 into `None` type(None). Defaults to `None` type(None), which means no parsing should occur.145 _env_parse_enums: Parse enum field names to values. Defaults to `None.`, which means no parsing should occur.146 _cli_prog_name: The CLI program name to display in help text. Defaults to `None` if _cli_parse_args is `None`.147 Otherwise, defaults to sys.argv[0].148 _cli_parse_args: The list of CLI arguments to parse. Defaults to None.149 If set to `True`, defaults to sys.argv[1:].150 _cli_settings_source: Override the default CLI settings source with a user defined instance. Defaults to None.151 _cli_parse_none_str: The CLI string value that should be parsed (e.g. "null", "void", "None", etc.) into152 `None` type(None). Defaults to _env_parse_none_str value if set. Otherwise, defaults to "null" if153 _cli_avoid_json is `False`, and "None" if _cli_avoid_json is `True`.154 _cli_hide_none_type: Hide `None` values in CLI help text. Defaults to `False`.155 _cli_avoid_json: Avoid complex JSON objects in CLI help text. Defaults to `False`.156 _cli_enforce_required: Enforce required fields at the CLI. Defaults to `False`.157 _cli_use_class_docs_for_groups: Use class docstrings in CLI group help text instead of field descriptions.158 Defaults to `False`.159 _cli_exit_on_error: Determines whether or not the internal parser exits with error info when an error occurs.160 Defaults to `True`.161 _cli_prefix: The root parser command line arguments prefix. Defaults to "".162 _cli_flag_prefix_char: The flag prefix character to use for CLI optional arguments. Defaults to '-'.163 _cli_implicit_flags: Controls how `bool` fields are exposed as CLI flags.164 165 - False (default): no implicit flags are generated; booleans must be set explicitly (e.g. --flag=true).166 - True / 'dual': optional boolean fields generate both positive and negative forms (--flag and --no-flag).167 - 'toggle': required boolean fields remain in 'dual' mode, while optional boolean fields generate a single168 flag aligned with the default value (if default=False, expose --flag; if default=True, expose --no-flag).169 _cli_ignore_unknown_args: Whether to ignore unknown CLI args and parse only known ones. Defaults to `False`.170 _cli_kebab_case: CLI args use kebab case. Defaults to `False`.171 _cli_shortcuts: Mapping of target field name to alias names. Defaults to `None`.172 _secrets_dir: The secret files directory or a sequence of directories. Defaults to `None`.173 _build_sources: Pre-initialized sources and init kwargs to use for building instantiation values.174 Defaults to `None`.175 """176 177 # Note: when adding new parameters, make sure to use `object` instead of `Any` to avoid issues with the Mypy plugin178 # when used with `--disallow-any-explicit`. If `Any` needs to be used as a generic parameter for variance (e.g. in `_build_sources`),179 # make sure to update the Pydantic Mypy plugin accordingly.180 def __init__(181 __pydantic_self__,182 _case_sensitive: bool | None = None,183 _nested_model_default_partial_update: bool | None = None,184 _env_prefix: str | None = None,185 _env_prefix_target: EnvPrefixTarget | None = None,186 _env_file: DotenvType | None = ENV_FILE_SENTINEL,187 _env_file_encoding: str | None = None,188 _env_ignore_empty: bool | None = None,189 _env_nested_delimiter: str | None = None,190 _env_nested_max_split: int | None = None,191 _env_parse_none_str: str | None = None,192 _env_parse_enums: bool | None = None,193 _cli_prog_name: str | None = None,194 _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None,195 _cli_settings_source: CliSettingsSource[Any] | None = None,196 _cli_parse_none_str: str | None = None,197 _cli_hide_none_type: bool | None = None,198 _cli_avoid_json: bool | None = None,199 _cli_enforce_required: bool | None = None,200 _cli_use_class_docs_for_groups: bool | None = None,201 _cli_exit_on_error: bool | None = None,202 _cli_prefix: str | None = None,203 _cli_flag_prefix_char: str | None = None,204 _cli_implicit_flags: bool | Literal['dual', 'toggle'] | None = None,205 _cli_ignore_unknown_args: bool | None = None,206 _cli_kebab_case: bool | Literal['all', 'no_enums'] | None = None,207 _cli_shortcuts: Mapping[str, str | list[str]] | None = None,208 _secrets_dir: PathType | None = None,209 _build_sources: tuple[tuple[PydanticBaseSettingsSource, ...], dict[str, Any]] | None = None,210 **values: Any,211 ) -> None:212 sources, init_kwargs = (213 _build_sources214 if _build_sources is not None215 else __pydantic_self__.__class__._settings_init_sources(216 _case_sensitive=_case_sensitive,217 _nested_model_default_partial_update=_nested_model_default_partial_update,218 _env_prefix=_env_prefix,219 _env_prefix_target=_env_prefix_target,220 _env_file=_env_file,221 _env_file_encoding=_env_file_encoding,222 _env_ignore_empty=_env_ignore_empty,223 _env_nested_delimiter=_env_nested_delimiter,224 _env_nested_max_split=_env_nested_max_split,225 _env_parse_none_str=_env_parse_none_str,226 _env_parse_enums=_env_parse_enums,227 _cli_prog_name=_cli_prog_name,228 _cli_parse_args=_cli_parse_args,229 _cli_settings_source=_cli_settings_source,230 _cli_parse_none_str=_cli_parse_none_str,231 _cli_hide_none_type=_cli_hide_none_type,232 _cli_avoid_json=_cli_avoid_json,233 _cli_enforce_required=_cli_enforce_required,234 _cli_use_class_docs_for_groups=_cli_use_class_docs_for_groups,235 _cli_exit_on_error=_cli_exit_on_error,236 _cli_prefix=_cli_prefix,237 _cli_flag_prefix_char=_cli_flag_prefix_char,238 _cli_implicit_flags=_cli_implicit_flags,239 _cli_ignore_unknown_args=_cli_ignore_unknown_args,240 _cli_kebab_case=_cli_kebab_case,241 _cli_shortcuts=_cli_shortcuts,242 _secrets_dir=_secrets_dir,243 _init_kwargs=values,244 )245 )246 247 super().__init__(**__pydantic_self__.__class__._settings_build_values(sources, init_kwargs))248 249 @classmethod250 def settings_customise_sources(251 cls,252 settings_cls: type[BaseSettings],253 init_settings: PydanticBaseSettingsSource,254 env_settings: PydanticBaseSettingsSource,255 dotenv_settings: PydanticBaseSettingsSource,256 file_secret_settings: PydanticBaseSettingsSource,257 ) -> tuple[PydanticBaseSettingsSource, ...]:258 """259 Define the sources and their order for loading the settings values.260 261 Args:262 settings_cls: The Settings class.263 init_settings: The `InitSettingsSource` instance.264 env_settings: The `EnvSettingsSource` instance.265 dotenv_settings: The `DotEnvSettingsSource` instance.266 file_secret_settings: The `SecretsSettingsSource` instance.267 268 Returns:269 A tuple containing the sources and their order for loading the settings values.270 """271 return init_settings, env_settings, dotenv_settings, file_secret_settings272 273 @classmethod274 def _settings_init_sources(275 cls,276 _case_sensitive: bool | None = None,277 _nested_model_default_partial_update: bool | None = None,278 _env_prefix: str | None = None,279 _env_prefix_target: EnvPrefixTarget | None = None,280 _env_file: DotenvType | None = ENV_FILE_SENTINEL,281 _env_file_encoding: str | None = None,282 _env_ignore_empty: bool | None = None,283 _env_nested_delimiter: str | None = None,284 _env_nested_max_split: int | None = None,285 _env_parse_none_str: str | None = None,286 _env_parse_enums: bool | None = None,287 _cli_prog_name: str | None = None,288 _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None,289 _cli_settings_source: CliSettingsSource[Any] | None = None,290 _cli_parse_none_str: str | None = None,291 _cli_hide_none_type: bool | None = None,292 _cli_avoid_json: bool | None = None,293 _cli_enforce_required: bool | None = None,294 _cli_use_class_docs_for_groups: bool | None = None,295 _cli_exit_on_error: bool | None = None,296 _cli_prefix: str | None = None,297 _cli_flag_prefix_char: str | None = None,298 _cli_implicit_flags: bool | Literal['dual', 'toggle'] | None = None,299 _cli_ignore_unknown_args: bool | None = None,300 _cli_kebab_case: bool | Literal['all', 'no_enums'] | None = None,301 _cli_shortcuts: Mapping[str, str | list[str]] | None = None,302 _secrets_dir: PathType | None = None,303 _init_kwargs: dict[str, Any] | None = None,304 ) -> tuple[tuple[PydanticBaseSettingsSource, ...], dict[str, Any]]:305 # Determine settings config values306 case_sensitive = _case_sensitive if _case_sensitive is not None else cls.model_config.get('case_sensitive')307 env_prefix = _env_prefix if _env_prefix is not None else cls.model_config.get('env_prefix')308 env_prefix_target = (309 _env_prefix_target if _env_prefix_target is not None else cls.model_config.get('env_prefix_target')310 )311 nested_model_default_partial_update = (312 _nested_model_default_partial_update313 if _nested_model_default_partial_update is not None314 else cls.model_config.get('nested_model_default_partial_update')315 )316 env_file = _env_file if _env_file != ENV_FILE_SENTINEL else cls.model_config.get('env_file')317 env_file_encoding = (318 _env_file_encoding if _env_file_encoding is not None else cls.model_config.get('env_file_encoding')319 )320 env_ignore_empty = (321 _env_ignore_empty if _env_ignore_empty is not None else cls.model_config.get('env_ignore_empty')322 )323 env_nested_delimiter = (324 _env_nested_delimiter if _env_nested_delimiter is not None else cls.model_config.get('env_nested_delimiter')325 )326 env_nested_max_split = (327 _env_nested_max_split if _env_nested_max_split is not None else cls.model_config.get('env_nested_max_split')328 )329 env_parse_none_str = (330 _env_parse_none_str if _env_parse_none_str is not None else cls.model_config.get('env_parse_none_str')331 )332 env_parse_enums = _env_parse_enums if _env_parse_enums is not None else cls.model_config.get('env_parse_enums')333 334 cli_prog_name = _cli_prog_name if _cli_prog_name is not None else cls.model_config.get('cli_prog_name')335 cli_parse_args = _cli_parse_args if _cli_parse_args is not None else cls.model_config.get('cli_parse_args')336 cli_settings_source = (337 _cli_settings_source if _cli_settings_source is not None else cls.model_config.get('cli_settings_source')338 )339 cli_parse_none_str = (340 _cli_parse_none_str if _cli_parse_none_str is not None else cls.model_config.get('cli_parse_none_str')341 )342 cli_parse_none_str = cli_parse_none_str if not env_parse_none_str else env_parse_none_str343 cli_hide_none_type = (344 _cli_hide_none_type if _cli_hide_none_type is not None else cls.model_config.get('cli_hide_none_type')345 )346 cli_avoid_json = _cli_avoid_json if _cli_avoid_json is not None else cls.model_config.get('cli_avoid_json')347 cli_enforce_required = (348 _cli_enforce_required if _cli_enforce_required is not None else cls.model_config.get('cli_enforce_required')349 )350 cli_use_class_docs_for_groups = (351 _cli_use_class_docs_for_groups352 if _cli_use_class_docs_for_groups is not None353 else cls.model_config.get('cli_use_class_docs_for_groups')354 )355 cli_exit_on_error = (356 _cli_exit_on_error if _cli_exit_on_error is not None else cls.model_config.get('cli_exit_on_error')357 )358 cli_prefix = _cli_prefix if _cli_prefix is not None else cls.model_config.get('cli_prefix')359 cli_flag_prefix_char = (360 _cli_flag_prefix_char if _cli_flag_prefix_char is not None else cls.model_config.get('cli_flag_prefix_char')361 )362 cli_implicit_flags = (363 _cli_implicit_flags if _cli_implicit_flags is not None else cls.model_config.get('cli_implicit_flags')364 )365 cli_ignore_unknown_args = (366 _cli_ignore_unknown_args367 if _cli_ignore_unknown_args is not None368 else cls.model_config.get('cli_ignore_unknown_args')369 )370 cli_kebab_case = _cli_kebab_case if _cli_kebab_case is not None else cls.model_config.get('cli_kebab_case')371 cli_shortcuts = _cli_shortcuts if _cli_shortcuts is not None else cls.model_config.get('cli_shortcuts')372 373 secrets_dir = _secrets_dir if _secrets_dir is not None else cls.model_config.get('secrets_dir')374 375 # Configure built-in sources376 default_settings = DefaultSettingsSource(377 cls, nested_model_default_partial_update=nested_model_default_partial_update378 )379 init_settings = InitSettingsSource(380 cls,381 init_kwargs=_init_kwargs if _init_kwargs is not None else {},382 nested_model_default_partial_update=nested_model_default_partial_update,383 )384 env_settings = EnvSettingsSource(385 cls,386 case_sensitive=case_sensitive,387 env_prefix=env_prefix,388 env_prefix_target=env_prefix_target,389 env_nested_delimiter=env_nested_delimiter,390 env_nested_max_split=env_nested_max_split,391 env_ignore_empty=env_ignore_empty,392 env_parse_none_str=env_parse_none_str,393 env_parse_enums=env_parse_enums,394 )395 dotenv_settings = DotEnvSettingsSource(396 cls,397 env_file=env_file,398 env_file_encoding=env_file_encoding,399 case_sensitive=case_sensitive,400 env_prefix=env_prefix,401 env_prefix_target=env_prefix_target,402 env_nested_delimiter=env_nested_delimiter,403 env_nested_max_split=env_nested_max_split,404 env_ignore_empty=env_ignore_empty,405 env_parse_none_str=env_parse_none_str,406 env_parse_enums=env_parse_enums,407 )408 409 file_secret_settings = SecretsSettingsSource(410 cls,411 secrets_dir=secrets_dir,412 case_sensitive=case_sensitive,413 env_prefix=env_prefix,414 env_prefix_target=env_prefix_target,415 )416 # Provide a hook to set built-in sources priority and add / remove sources417 sources = cls.settings_customise_sources(418 cls,419 init_settings=init_settings,420 env_settings=env_settings,421 dotenv_settings=dotenv_settings,422 file_secret_settings=file_secret_settings,423 ) + (default_settings,)424 custom_cli_sources = [source for source in sources if isinstance(source, CliSettingsSource)]425 if not any(custom_cli_sources):426 if isinstance(cli_settings_source, CliSettingsSource):427 sources = (cli_settings_source,) + sources428 elif cli_parse_args is not None:429 cli_settings = CliSettingsSource[Any](430 cls,431 cli_prog_name=cli_prog_name,432 cli_parse_args=cli_parse_args,433 cli_parse_none_str=cli_parse_none_str,434 cli_hide_none_type=cli_hide_none_type,435 cli_avoid_json=cli_avoid_json,436 cli_enforce_required=cli_enforce_required,437 cli_use_class_docs_for_groups=cli_use_class_docs_for_groups,438 cli_exit_on_error=cli_exit_on_error,439 cli_prefix=cli_prefix,440 cli_flag_prefix_char=cli_flag_prefix_char,441 cli_implicit_flags=cli_implicit_flags,442 cli_ignore_unknown_args=cli_ignore_unknown_args,443 cli_kebab_case=cli_kebab_case,444 cli_shortcuts=cli_shortcuts,445 case_sensitive=case_sensitive,446 )447 sources = (cli_settings,) + sources448 # We ensure that if command line arguments haven't been parsed yet, we do so.449 elif cli_parse_args not in (None, False) and not custom_cli_sources[0].env_vars:450 custom_cli_sources[0](args=cli_parse_args) # type: ignore451 452 cls._settings_warn_unused_config_keys(sources, cls.model_config)453 454 return sources, _init_kwargs if _init_kwargs is not None else {}455 456 @classmethod457 def _settings_build_values(458 cls, sources: tuple[PydanticBaseSettingsSource, ...], init_kwargs: dict[str, Any]459 ) -> dict[str, Any]:460 if sources:461 state: dict[str, Any] = {}462 defaults: dict[str, Any] = {}463 states: dict[str, dict[str, Any]] = {}464 for source in sources:465 if isinstance(source, PydanticBaseSettingsSource):466 source._set_current_state(state)467 source._set_settings_sources_data(states)468 469 source_name = source.__name__ if hasattr(source, '__name__') else type(source).__name__470 source_state = source()471 472 if isinstance(source, DefaultSettingsSource):473 defaults = source_state474 475 states[source_name] = source_state476 state = deep_update(source_state, state)477 478 # Strip any default values not explicity set before returning final state479 state = {key: val for key, val in state.items() if key not in defaults or defaults[key] != val}480 cls._settings_restore_init_kwarg_names(cls, init_kwargs, state)481 482 return state483 else:484 # no one should mean to do this, but I think returning an empty dict is marginally preferable485 # to an informative error and much better than a confusing error486 return {}487 488 @staticmethod489 def _settings_restore_init_kwarg_names(490 settings_cls: type[BaseSettings], init_kwargs: dict[str, Any], state: dict[str, Any]491 ) -> None:492 """493 Restore the init_kwarg key names to the final merged state dictionary.494 495 This function renames keys in state to match the original init_kwargs key names,496 preserving the merged values from the source priority order.497 """498 if init_kwargs and state:499 state_kwarg_names = set(state.keys())500 init_kwarg_names = set(init_kwargs.keys())501 for field_name, field_info in settings_cls.model_fields.items():502 alias_names, *_ = _get_alias_names(field_name, field_info)503 matchable_names = set(alias_names)504 include_name = settings_cls.model_config.get(505 'populate_by_name', False506 ) or settings_cls.model_config.get('validate_by_name', False)507 if include_name:508 matchable_names.add(field_name)509 init_kwarg_name = init_kwarg_names & matchable_names510 state_kwarg_name = state_kwarg_names & matchable_names511 if init_kwarg_name and state_kwarg_name:512 # Use deterministic selection for both keys.513 # Target key: the key from init_kwargs that should be used in the final state.514 target_key = next(iter(init_kwarg_name))515 # Source key: prefer the alias (first in alias_names) if present in state,516 # as InitSettingsSource normalizes to the preferred alias.517 # This ensures we get the highest-priority value for this field.518 source_key = None519 for alias in alias_names:520 if alias in state_kwarg_name:521 source_key = alias522 break523 if source_key is None:524 # Fall back to field_name if no alias found in state525 source_key = field_name if field_name in state_kwarg_name else next(iter(state_kwarg_name))526 # Get the value from the source key and remove all matching keys527 value = state.pop(source_key)528 for key in state_kwarg_name - {source_key}:529 state.pop(key, None)530 state[target_key] = value531 532 @staticmethod533 def _settings_warn_unused_config_keys(sources: tuple[object, ...], model_config: SettingsConfigDict) -> None:534 """535 Warns if any values in model_config were set but the corresponding settings source has not been initialised.536 537 The list alternative sources and their config keys can be found here:538 https://docs.pydantic.dev/latest/concepts/pydantic_settings/#other-settings-source539 540 Args:541 sources: The tuple of configured sources542 model_config: The model config to check for unused config keys543 """544 545 def warn_if_not_used(source_type: type[PydanticBaseSettingsSource], keys: tuple[str, ...]) -> None:546 if not any(isinstance(source, source_type) for source in sources):547 for key in keys:548 if model_config.get(key) is not None:549 warnings.warn(550 f'Config key `{key}` is set in model_config but will be ignored because no '551 f'{source_type.__name__} source is configured. To use this config key, add a '552 f'{source_type.__name__} source to the settings sources via the '553 'settings_customise_sources hook.',554 UserWarning,555 stacklevel=3,556 )557 558 warn_if_not_used(JsonConfigSettingsSource, ('json_file', 'json_file_encoding'))559 warn_if_not_used(PyprojectTomlConfigSettingsSource, ('pyproject_toml_depth', 'pyproject_toml_table_header'))560 warn_if_not_used(TomlConfigSettingsSource, ('toml_file',))561 warn_if_not_used(YamlConfigSettingsSource, ('yaml_file', 'yaml_file_encoding', 'yaml_config_section'))562 563 model_config: ClassVar[SettingsConfigDict] = SettingsConfigDict(564 extra='forbid',565 arbitrary_types_allowed=True,566 validate_default=True,567 case_sensitive=False,568 env_prefix='',569 env_prefix_target='variable',570 nested_model_default_partial_update=False,571 env_file=None,572 env_file_encoding=None,573 env_ignore_empty=False,574 env_nested_delimiter=None,575 env_nested_max_split=None,576 env_parse_none_str=None,577 env_parse_enums=None,578 cli_prog_name=None,579 cli_parse_args=None,580 cli_parse_none_str=None,581 cli_hide_none_type=False,582 cli_avoid_json=False,583 cli_enforce_required=False,584 cli_use_class_docs_for_groups=False,585 cli_exit_on_error=True,586 cli_prefix='',587 cli_flag_prefix_char='-',588 cli_implicit_flags=False,589 cli_ignore_unknown_args=False,590 cli_kebab_case=False,591 cli_shortcuts=None,592 json_file=None,593 json_file_encoding=None,594 yaml_file=None,595 yaml_file_encoding=None,596 yaml_config_section=None,597 toml_file=None,598 secrets_dir=None,599 protected_namespaces=('model_validate', 'model_dump', 'settings_customise_sources'),600 enable_decoding=True,601 )602 603 604class CliApp:605 """606 A utility class for running Pydantic `BaseSettings`, `BaseModel`, or `pydantic.dataclasses.dataclass` as607 CLI applications.608 """609 610 _subcommand_stack: ClassVar[dict[int, tuple[CliSettingsSource[Any], Any, str]]] = {}611 _ansi_color: ClassVar[re.Pattern[str]] = re.compile(r'\x1b\[[0-9;]*m')612 613 @staticmethod614 def _get_base_settings_cls(model_cls: type[Any]) -> type[BaseSettings]:615 if issubclass(model_cls, BaseSettings):616 return model_cls617 618 class CliAppBaseSettings(BaseSettings, model_cls): # type: ignore619 __doc__ = model_cls.__doc__620 model_config = SettingsConfigDict(621 nested_model_default_partial_update=True,622 case_sensitive=True,623 cli_hide_none_type=True,624 cli_avoid_json=True,625 cli_enforce_required=True,626 cli_implicit_flags=True,627 cli_kebab_case=True,628 )629 630 return CliAppBaseSettings631 632 @staticmethod633 def _run_cli_cmd(model: Any, cli_cmd_method_name: str, is_required: bool) -> Any:634 command = getattr(type(model), cli_cmd_method_name, None)635 if command is None:636 if is_required:637 raise SettingsError(f'Error: {type(model).__name__} class is missing {cli_cmd_method_name} entrypoint')638 return model639 640 # If the method is asynchronous, we handle its execution based on the current event loop status.641 if inspect.iscoroutinefunction(command):642 # For asynchronous methods, we have two execution scenarios:643 # 1. If no event loop is running in the current thread, run the coroutine directly with asyncio.run().644 # 2. If an event loop is already running in the current thread, run the coroutine in a separate thread to avoid conflicts.645 try:646 # Check if an event loop is currently running in this thread.647 loop = asyncio.get_running_loop()648 except RuntimeError:649 loop = None650 651 if loop and loop.is_running():652 # We're in a context with an active event loop (e.g., Jupyter Notebook).653 # Running asyncio.run() here would cause conflicts, so we use a separate thread.654 exception_container = []655 656 def run_coro() -> None:657 try:658 # Execute the coroutine in a new event loop in this separate thread.659 asyncio.run(command(model))660 except Exception as e:661 exception_container.append(e)662 663 thread = threading.Thread(target=run_coro)664 thread.start()665 thread.join()666 if exception_container:667 # Propagate exceptions from the separate thread.668 raise exception_container[0]669 else:670 # No event loop is running; safe to run the coroutine directly.671 asyncio.run(command(model))672 else:673 # For synchronous methods, call them directly.674 command(model)675 676 return model677 678 @staticmethod679 def run(680 model_cls: type[T],681 cli_args: list[str] | Namespace | SimpleNamespace | dict[str, Any] | None = None,682 cli_settings_source: CliSettingsSource[Any] | None = None,683 cli_exit_on_error: bool | None = None,684 cli_cmd_method_name: str = 'cli_cmd',685 **model_init_data: Any,686 ) -> T:687 """688 Runs a Pydantic `BaseSettings`, `BaseModel`, or `pydantic.dataclasses.dataclass` as a CLI application.689 Running a model as a CLI application requires the `cli_cmd` method to be defined in the model class.690 691 Args:692 model_cls: The model class to run as a CLI application.693 cli_args: The list of CLI arguments to parse. If `cli_settings_source` is specified, this may694 also be a namespace or dictionary of pre-parsed CLI arguments. Defaults to `sys.argv[1:]`.695 cli_settings_source: Override the default CLI settings source with a user defined instance.696 Defaults to `None`.697 cli_exit_on_error: Determines whether this function exits on error. If model is subclass of698 `BaseSettings`, defaults to BaseSettings `cli_exit_on_error` value. Otherwise, defaults to699 `True`.700 cli_cmd_method_name: The CLI command method name to run. Defaults to "cli_cmd".701 model_init_data: The model init data.702 703 Returns:704 The ran instance of model.705 706 Raises:707 SettingsError: If model_cls is not subclass of `BaseModel` or `pydantic.dataclasses.dataclass`.708 SettingsError: If model_cls does not have a `cli_cmd` entrypoint defined.709 """710 711 if not (is_pydantic_dataclass(model_cls) or is_model_class(model_cls)):712 raise SettingsError(713 f'Error: {model_cls.__name__} is not subclass of BaseModel or pydantic.dataclasses.dataclass'714 )715 716 cli_settings = None717 cli_parse_args = True if cli_args is None else cli_args718 if cli_settings_source is not None:719 if isinstance(cli_parse_args, (Namespace, SimpleNamespace, dict)):720 cli_settings = cli_settings_source(parsed_args=cli_parse_args)721 else:722 cli_settings = cli_settings_source(args=cli_parse_args)723 elif isinstance(cli_parse_args, (Namespace, SimpleNamespace, dict)):724 raise SettingsError('Error: `cli_args` must be list[str] or None when `cli_settings_source` is not used')725 726 if not issubclass(model_cls, BaseSettings):727 base_settings_cls = CliApp._get_base_settings_cls(model_cls)728 sources, init_kwargs = base_settings_cls._settings_init_sources(729 _cli_parse_args=cli_parse_args, # type: ignore[arg-type]730 _cli_exit_on_error=cli_exit_on_error,731 _cli_settings_source=cli_settings,732 _init_kwargs=model_init_data,733 )734 model = base_settings_cls(**base_settings_cls._settings_build_values(sources, init_kwargs))735 model_init_data = {}736 for field_name, field_info in base_settings_cls.model_fields.items():737 model_init_data[_field_name_for_signature(field_name, field_info)] = getattr(model, field_name)738 command = model_cls(**model_init_data)739 else:740 sources, init_kwargs = model_cls._settings_init_sources(741 _cli_parse_args=cli_parse_args, # type: ignore[arg-type]742 _cli_exit_on_error=cli_exit_on_error,743 _cli_settings_source=cli_settings,744 _init_kwargs=model_init_data,745 )746 command = model_cls(_build_sources=(sources, init_kwargs))747 748 subcommand_dest = ':subcommand'749 cli_settings_source = [source for source in sources if isinstance(source, CliSettingsSource)][0]750 CliApp._subcommand_stack[id(command)] = (cli_settings_source, cli_settings_source.root_parser, subcommand_dest)751 try:752 data_model = CliApp._run_cli_cmd(command, cli_cmd_method_name, is_required=False)753 finally:754 del CliApp._subcommand_stack[id(command)]755 return data_model756 757 @staticmethod758 def run_subcommand(759 model: PydanticModel, cli_exit_on_error: bool | None = None, cli_cmd_method_name: str = 'cli_cmd'760 ) -> PydanticModel:761 """762 Runs the model subcommand. Running a model subcommand requires the `cli_cmd` method to be defined in763 the nested model subcommand class.764 765 Args:766 model: The model to run the subcommand from.767 cli_exit_on_error: Determines whether this function exits with error if no subcommand is found.768 Defaults to model_config `cli_exit_on_error` value if set. Otherwise, defaults to `True`.769 cli_cmd_method_name: The CLI command method name to run. Defaults to "cli_cmd".770 771 Returns:772 The ran subcommand model.773 774 Raises:775 SystemExit: When no subcommand is found and cli_exit_on_error=`True` (the default).776 SettingsError: When no subcommand is found and cli_exit_on_error=`False`.777 """778 779 if id(model) in CliApp._subcommand_stack:780 cli_settings_source, parser, subcommand_dest = CliApp._subcommand_stack[id(model)]781 else:782 cli_settings_source = CliSettingsSource[Any](CliApp._get_base_settings_cls(type(model)))783 parser = cli_settings_source.root_parser784 subcommand_dest = ':subcommand'785 786 cli_exit_on_error = cli_settings_source.cli_exit_on_error if cli_exit_on_error is None else cli_exit_on_error787 788 errors: list[SettingsError | SystemExit] = []789 subcommand = get_subcommand(790 model, is_required=True, cli_exit_on_error=cli_exit_on_error, _suppress_errors=errors791 )792 if errors:793 err = errors[0]794 if err.__context__ is None and err.__cause__ is None and cli_settings_source._format_help is not None:795 error_message = f'{err}\n{cli_settings_source._format_help(parser)}'796 raise type(err)(error_message) from None797 else:798 raise err799 800 subcommand_cls = cast(type[BaseModel], type(subcommand))801 subcommand_arg = cli_settings_source._parser_map[subcommand_dest][subcommand_cls]802 subcommand_dest = f'{subcommand_arg.dest}.:subcommand'803 subcommand_parser = subcommand_arg.parser804 CliApp._subcommand_stack[id(subcommand)] = (cli_settings_source, subcommand_parser, subcommand_dest)805 try:806 data_model = CliApp._run_cli_cmd(subcommand, cli_cmd_method_name, is_required=True)807 finally:808 del CliApp._subcommand_stack[id(subcommand)]809 return data_model810 811 @staticmethod812 def serialize(813 model: PydanticModel,814 list_style: Literal['json', 'argparse', 'lazy'] = 'json',815 dict_style: Literal['json', 'env'] = 'json',816 positionals_first: bool = False,817 ) -> list[str]:818 """819 Serializes the CLI arguments for a Pydantic data model.820 821 Args:822 model: The data model to serialize.823 list_style:824 Controls how list-valued fields are serialized on the command line.825 - 'json' (default):826 Lists are encoded as a single JSON array.827 Example: `--tags '["a","b","c"]'`828 - 'argparse':829 Each list element becomes its own repeated flag, following830 typical `argparse` conventions.831 Example: `--tags a --tags b --tags c`832 - 'lazy':833 Lists are emitted as a single comma-separated string without JSON834 quoting or escaping.835 Example: `--tags a,b,c`836 dict_style:837 Controls how dictionary-valued fields are serialized.838 - 'json' (default):839 The entire dictionary is emitted as a single JSON object.840 Example: `--config '{"host": "localhost", "port": 5432}'`841 - 'env':842 The dictionary is flattened into multiple CLI flags using843 environment-variable-style assignement.844 Example: `--config host=localhost --config port=5432`845 positionals_first: Controls whether positional arguments should be serialized846 first compared to optional arguments. Defaults to `False`.847 848 Returns:849 The serialized CLI arguments for the data model.850 """851 852 base_settings_cls = CliApp._get_base_settings_cls(type(model))853 serialized_args = CliSettingsSource[Any](base_settings_cls)._serialized_args(854 model,855 list_style=list_style,856 dict_style=dict_style,857 positionals_first=positionals_first,858 )859 return CliSettingsSource._flatten_serialized_args(serialized_args, positionals_first)860 861 @staticmethod862 def format_help(863 model: PydanticModel | type[T],864 cli_settings_source: CliSettingsSource[Any] | None = None,865 strip_ansi_color: bool = False,866 ) -> str:867 """868 Return a string containing a help message for a Pydantic model.869 870 Args:871 model: The model or model class.872 cli_settings_source: Override the default CLI settings source with a user defined instance.873 Defaults to `None`.874 strip_ansi_color: Strips ANSI color codes from the help message when set to `True`.875 876 Returns:877 The help message string for the model.878 """879 model_cls = model if isinstance(model, type) else type(model)880 if cli_settings_source is None:881 if not isinstance(model, type) and id(model) in CliApp._subcommand_stack:882 cli_settings_source, *_ = CliApp._subcommand_stack[id(model)]883 else:884 cli_settings_source = CliSettingsSource(CliApp._get_base_settings_cls(model_cls))885 help_message = cli_settings_source._format_help(cli_settings_source.root_parser)886 return help_message if not strip_ansi_color else CliApp._ansi_color.sub('', help_message)887 888 @staticmethod889 def print_help(890 model: PydanticModel | type[T],891 cli_settings_source: CliSettingsSource[Any] | None = None,892 file: TextIO | None = None,893 strip_ansi_color: bool = False,894 ) -> None:895 """896 Print a help message for a Pydantic model.897 898 Args:899 model: The model or model class.900 cli_settings_source: Override the default CLI settings source with a user defined instance.901 Defaults to `None`.902 file: A text stream to which the help message is written. If `None`, the output is sent to sys.stdout.903 strip_ansi_color: Strips ANSI color codes from the help message when set to `True`.904 """905 print(906 CliApp.format_help(907 model,908 cli_settings_source=cli_settings_source,909 strip_ansi_color=strip_ansi_color,910 ),911 file=file,912 )913 