from__future__importannotationsas_annotationsimportasyncioimportinspectimportthreadingfromargparseimportNamespacefromtypesimportSimpleNamespacefromtypingimportAny,ClassVar,TypeVarfrompydanticimportConfigDictfrompydantic._internal._configimportconfig_keysfrompydantic._internal._signatureimport_field_name_for_signaturefrompydantic._internal._utilsimportdeep_update,is_model_classfrompydantic.dataclassesimportis_pydantic_dataclassfrompydantic.mainimportBaseModelfrom.sourcesimport(ENV_FILE_SENTINEL,CliSettingsSource,DefaultSettingsSource,DotEnvSettingsSource,DotenvType,EnvSettingsSource,InitSettingsSource,PathType,PydanticBaseSettingsSource,PydanticModel,SecretsSettingsSource,SettingsError,get_subcommand,)T=TypeVar('T')classSettingsConfigDict(ConfigDict,total=False):case_sensitive:boolnested_model_default_partial_update:bool|Noneenv_prefix:strenv_file:DotenvType|Noneenv_file_encoding:str|Noneenv_ignore_empty:boolenv_nested_delimiter:str|Noneenv_nested_max_split:int|Noneenv_parse_none_str:str|Noneenv_parse_enums:bool|Nonecli_prog_name:str|Nonecli_parse_args:bool|list[str]|tuple[str,...]|Nonecli_parse_none_str:str|Nonecli_hide_none_type:boolcli_avoid_json:boolcli_enforce_required:boolcli_use_class_docs_for_groups:boolcli_exit_on_error:boolcli_prefix:strcli_flag_prefix_char:strcli_implicit_flags:bool|Nonecli_ignore_unknown_args:bool|Nonecli_kebab_case:bool|Nonesecrets_dir:PathType|Nonejson_file:PathType|Nonejson_file_encoding:str|Noneyaml_file:PathType|Noneyaml_file_encoding:str|Nonepyproject_toml_depth:int""" Number of levels **up** from the current working directory to attempt to find a pyproject.toml file. This is only used when a pyproject.toml file is not found in the current working directory. """pyproject_toml_table_header:tuple[str,...]""" Header of the TOML table within a pyproject.toml file to use when filling variables. This is supplied as a `tuple[str, ...]` instead of a `str` to accommodate for headers containing a `.`. For example, `toml_table_header = ("tool", "my.tool", "foo")` can be used to fill variable values from a table with header `[tool."my.tool".foo]`. To use the root table, exclude this config setting or provide an empty tuple. """toml_file:PathType|Noneenable_decoding:bool# Extend `config_keys` by pydantic settings config keys to# support setting config through class kwargs.# Pydantic uses `config_keys` in `pydantic._internal._config.ConfigWrapper.for_model`# to extract config keys from model kwargs, So, by adding pydantic settings keys to# `config_keys`, they will be considered as valid config keys and will be collected# by Pydantic.config_keys|=set(SettingsConfigDict.__annotations__.keys())classBaseSettings(BaseModel):""" Base class for settings, allowing values to be overridden by environment variables. This is useful in production for secrets you do not wish to save in code, it plays nicely with docker(-compose), Heroku and any 12 factor app design. All the below attributes can be set via `model_config`. Args: _case_sensitive: Whether environment and CLI variable names should be read with case-sensitivity. Defaults to `None`. _nested_model_default_partial_update: Whether to allow partial updates on nested model default object fields. Defaults to `False`. _env_prefix: Prefix for all environment variables. Defaults to `None`. _env_file: The env file(s) to load settings values from. Defaults to `Path('')`, which means that the value from `model_config['env_file']` should be used. You can also pass `None` to indicate that environment variables should not be loaded from an env file. _env_file_encoding: The env file encoding, e.g. `'latin-1'`. Defaults to `None`. _env_ignore_empty: Ignore environment variables where the value is an empty string. Default to `False`. _env_nested_delimiter: The nested env values delimiter. Defaults to `None`. _env_nested_max_split: The nested env values maximum nesting. Defaults to `None`, which means no limit. _env_parse_none_str: The env string value that should be parsed (e.g. "null", "void", "None", etc.) into `None` type(None). Defaults to `None` type(None), which means no parsing should occur. _env_parse_enums: Parse enum field names to values. Defaults to `None.`, which means no parsing should occur. _cli_prog_name: The CLI program name to display in help text. Defaults to `None` if _cli_parse_args is `None`. Otherwse, defaults to sys.argv[0]. _cli_parse_args: The list of CLI arguments to parse. Defaults to None. If set to `True`, defaults to sys.argv[1:]. _cli_settings_source: Override the default CLI settings source with a user defined instance. Defaults to None. _cli_parse_none_str: The CLI string value that should be parsed (e.g. "null", "void", "None", etc.) into `None` type(None). Defaults to _env_parse_none_str value if set. Otherwise, defaults to "null" if _cli_avoid_json is `False`, and "None" if _cli_avoid_json is `True`. _cli_hide_none_type: Hide `None` values in CLI help text. Defaults to `False`. _cli_avoid_json: Avoid complex JSON objects in CLI help text. Defaults to `False`. _cli_enforce_required: Enforce required fields at the CLI. Defaults to `False`. _cli_use_class_docs_for_groups: Use class docstrings in CLI group help text instead of field descriptions. Defaults to `False`. _cli_exit_on_error: Determines whether or not the internal parser exits with error info when an error occurs. Defaults to `True`. _cli_prefix: The root parser command line arguments prefix. Defaults to "". _cli_flag_prefix_char: The flag prefix character to use for CLI optional arguments. Defaults to '-'. _cli_implicit_flags: Whether `bool` fields should be implicitly converted into CLI boolean flags. (e.g. --flag, --no-flag). Defaults to `False`. _cli_ignore_unknown_args: Whether to ignore unknown CLI args and parse only known ones. Defaults to `False`. _cli_kebab_case: CLI args use kebab case. Defaults to `False`. _secrets_dir: The secret files directory or a sequence of directories. Defaults to `None`. """def__init__(__pydantic_self__,_case_sensitive:bool|None=None,_nested_model_default_partial_update:bool|None=None,_env_prefix:str|None=None,_env_file:DotenvType|None=ENV_FILE_SENTINEL,_env_file_encoding:str|None=None,_env_ignore_empty:bool|None=None,_env_nested_delimiter:str|None=None,_env_nested_max_split:int|None=None,_env_parse_none_str:str|None=None,_env_parse_enums:bool|None=None,_cli_prog_name:str|None=None,_cli_parse_args:bool|list[str]|tuple[str,...]|None=None,_cli_settings_source:CliSettingsSource[Any]|None=None,_cli_parse_none_str:str|None=None,_cli_hide_none_type:bool|None=None,_cli_avoid_json:bool|None=None,_cli_enforce_required:bool|None=None,_cli_use_class_docs_for_groups:bool|None=None,_cli_exit_on_error:bool|None=None,_cli_prefix:str|None=None,_cli_flag_prefix_char:str|None=None,_cli_implicit_flags:bool|None=None,_cli_ignore_unknown_args:bool|None=None,_cli_kebab_case:bool|None=None,_secrets_dir:PathType|None=None,**values:Any,)->None:super().__init__(**__pydantic_self__._settings_build_values(values,_case_sensitive=_case_sensitive,_nested_model_default_partial_update=_nested_model_default_partial_update,_env_prefix=_env_prefix,_env_file=_env_file,_env_file_encoding=_env_file_encoding,_env_ignore_empty=_env_ignore_empty,_env_nested_delimiter=_env_nested_delimiter,_env_nested_max_split=_env_nested_max_split,_env_parse_none_str=_env_parse_none_str,_env_parse_enums=_env_parse_enums,_cli_prog_name=_cli_prog_name,_cli_parse_args=_cli_parse_args,_cli_settings_source=_cli_settings_source,_cli_parse_none_str=_cli_parse_none_str,_cli_hide_none_type=_cli_hide_none_type,_cli_avoid_json=_cli_avoid_json,_cli_enforce_required=_cli_enforce_required,_cli_use_class_docs_for_groups=_cli_use_class_docs_for_groups,_cli_exit_on_error=_cli_exit_on_error,_cli_prefix=_cli_prefix,_cli_flag_prefix_char=_cli_flag_prefix_char,_cli_implicit_flags=_cli_implicit_flags,_cli_ignore_unknown_args=_cli_ignore_unknown_args,_cli_kebab_case=_cli_kebab_case,_secrets_dir=_secrets_dir,))@classmethoddefsettings_customise_sources(cls,settings_cls:type[BaseSettings],init_settings:PydanticBaseSettingsSource,env_settings:PydanticBaseSettingsSource,dotenv_settings:PydanticBaseSettingsSource,file_secret_settings:PydanticBaseSettingsSource,)->tuple[PydanticBaseSettingsSource,...]:""" Define the sources and their order for loading the settings values. Args: settings_cls: The Settings class. init_settings: The `InitSettingsSource` instance. env_settings: The `EnvSettingsSource` instance. dotenv_settings: The `DotEnvSettingsSource` instance. file_secret_settings: The `SecretsSettingsSource` instance. Returns: A tuple containing the sources and their order for loading the settings values. """returninit_settings,env_settings,dotenv_settings,file_secret_settingsdef_settings_build_values(self,init_kwargs:dict[str,Any],_case_sensitive:bool|None=None,_nested_model_default_partial_update:bool|None=None,_env_prefix:str|None=None,_env_file:DotenvType|None=None,_env_file_encoding:str|None=None,_env_ignore_empty:bool|None=None,_env_nested_delimiter:str|None=None,_env_nested_max_split:int|None=None,_env_parse_none_str:str|None=None,_env_parse_enums:bool|None=None,_cli_prog_name:str|None=None,_cli_parse_args:bool|list[str]|tuple[str,...]|None=None,_cli_settings_source:CliSettingsSource[Any]|None=None,_cli_parse_none_str:str|None=None,_cli_hide_none_type:bool|None=None,_cli_avoid_json:bool|None=None,_cli_enforce_required:bool|None=None,_cli_use_class_docs_for_groups:bool|None=None,_cli_exit_on_error:bool|None=None,_cli_prefix:str|None=None,_cli_flag_prefix_char:str|None=None,_cli_implicit_flags:bool|None=None,_cli_ignore_unknown_args:bool|None=None,_cli_kebab_case:bool|None=None,_secrets_dir:PathType|None=None,)->dict[str,Any]:# Determine settings config valuescase_sensitive=_case_sensitiveif_case_sensitiveisnotNoneelseself.model_config.get('case_sensitive')env_prefix=_env_prefixif_env_prefixisnotNoneelseself.model_config.get('env_prefix')nested_model_default_partial_update=(_nested_model_default_partial_updateif_nested_model_default_partial_updateisnotNoneelseself.model_config.get('nested_model_default_partial_update'))env_file=_env_fileif_env_file!=ENV_FILE_SENTINELelseself.model_config.get('env_file')env_file_encoding=(_env_file_encodingif_env_file_encodingisnotNoneelseself.model_config.get('env_file_encoding'))env_ignore_empty=(_env_ignore_emptyif_env_ignore_emptyisnotNoneelseself.model_config.get('env_ignore_empty'))env_nested_delimiter=(_env_nested_delimiterif_env_nested_delimiterisnotNoneelseself.model_config.get('env_nested_delimiter'))env_nested_max_split=(_env_nested_max_splitif_env_nested_max_splitisnotNoneelseself.model_config.get('env_nested_max_split'))env_parse_none_str=(_env_parse_none_strif_env_parse_none_strisnotNoneelseself.model_config.get('env_parse_none_str'))env_parse_enums=_env_parse_enumsif_env_parse_enumsisnotNoneelseself.model_config.get('env_parse_enums')cli_prog_name=_cli_prog_nameif_cli_prog_nameisnotNoneelseself.model_config.get('cli_prog_name')cli_parse_args=_cli_parse_argsif_cli_parse_argsisnotNoneelseself.model_config.get('cli_parse_args')cli_settings_source=(_cli_settings_sourceif_cli_settings_sourceisnotNoneelseself.model_config.get('cli_settings_source'))cli_parse_none_str=(_cli_parse_none_strif_cli_parse_none_strisnotNoneelseself.model_config.get('cli_parse_none_str'))cli_parse_none_str=cli_parse_none_strifnotenv_parse_none_strelseenv_parse_none_strcli_hide_none_type=(_cli_hide_none_typeif_cli_hide_none_typeisnotNoneelseself.model_config.get('cli_hide_none_type'))cli_avoid_json=_cli_avoid_jsonif_cli_avoid_jsonisnotNoneelseself.model_config.get('cli_avoid_json')cli_enforce_required=(_cli_enforce_requiredif_cli_enforce_requiredisnotNoneelseself.model_config.get('cli_enforce_required'))cli_use_class_docs_for_groups=(_cli_use_class_docs_for_groupsif_cli_use_class_docs_for_groupsisnotNoneelseself.model_config.get('cli_use_class_docs_for_groups'))cli_exit_on_error=(_cli_exit_on_errorif_cli_exit_on_errorisnotNoneelseself.model_config.get('cli_exit_on_error'))cli_prefix=_cli_prefixif_cli_prefixisnotNoneelseself.model_config.get('cli_prefix')cli_flag_prefix_char=(_cli_flag_prefix_charif_cli_flag_prefix_charisnotNoneelseself.model_config.get('cli_flag_prefix_char'))cli_implicit_flags=(_cli_implicit_flagsif_cli_implicit_flagsisnotNoneelseself.model_config.get('cli_implicit_flags'))cli_ignore_unknown_args=(_cli_ignore_unknown_argsif_cli_ignore_unknown_argsisnotNoneelseself.model_config.get('cli_ignore_unknown_args'))cli_kebab_case=_cli_kebab_caseif_cli_kebab_caseisnotNoneelseself.model_config.get('cli_kebab_case')secrets_dir=_secrets_dirif_secrets_dirisnotNoneelseself.model_config.get('secrets_dir')# Configure built-in sourcesdefault_settings=DefaultSettingsSource(self.__class__,nested_model_default_partial_update=nested_model_default_partial_update)init_settings=InitSettingsSource(self.__class__,init_kwargs=init_kwargs,nested_model_default_partial_update=nested_model_default_partial_update,)env_settings=EnvSettingsSource(self.__class__,case_sensitive=case_sensitive,env_prefix=env_prefix,env_nested_delimiter=env_nested_delimiter,env_nested_max_split=env_nested_max_split,env_ignore_empty=env_ignore_empty,env_parse_none_str=env_parse_none_str,env_parse_enums=env_parse_enums,)dotenv_settings=DotEnvSettingsSource(self.__class__,env_file=env_file,env_file_encoding=env_file_encoding,case_sensitive=case_sensitive,env_prefix=env_prefix,env_nested_delimiter=env_nested_delimiter,env_nested_max_split=env_nested_max_split,env_ignore_empty=env_ignore_empty,env_parse_none_str=env_parse_none_str,env_parse_enums=env_parse_enums,)file_secret_settings=SecretsSettingsSource(self.__class__,secrets_dir=secrets_dir,case_sensitive=case_sensitive,env_prefix=env_prefix)# Provide a hook to set built-in sources priority and add / remove sourcessources=self.settings_customise_sources(self.__class__,init_settings=init_settings,env_settings=env_settings,dotenv_settings=dotenv_settings,file_secret_settings=file_secret_settings,)+(default_settings,)ifnotany([sourceforsourceinsourcesifisinstance(source,CliSettingsSource)]):ifisinstance(cli_settings_source,CliSettingsSource):sources=(cli_settings_source,)+sourceselifcli_parse_argsisnotNone:cli_settings=CliSettingsSource[Any](self.__class__,cli_prog_name=cli_prog_name,cli_parse_args=cli_parse_args,cli_parse_none_str=cli_parse_none_str,cli_hide_none_type=cli_hide_none_type,cli_avoid_json=cli_avoid_json,cli_enforce_required=cli_enforce_required,cli_use_class_docs_for_groups=cli_use_class_docs_for_groups,cli_exit_on_error=cli_exit_on_error,cli_prefix=cli_prefix,cli_flag_prefix_char=cli_flag_prefix_char,cli_implicit_flags=cli_implicit_flags,cli_ignore_unknown_args=cli_ignore_unknown_args,cli_kebab_case=cli_kebab_case,case_sensitive=case_sensitive,)sources=(cli_settings,)+sourcesifsources:state:dict[str,Any]={}states:dict[str,dict[str,Any]]={}forsourceinsources:ifisinstance(source,PydanticBaseSettingsSource):source._set_current_state(state)source._set_settings_sources_data(states)source_name=source.__name__ifhasattr(source,'__name__')elsetype(source).__name__source_state=source()states[source_name]=source_statestate=deep_update(source_state,state)returnstateelse:# no one should mean to do this, but I think returning an empty dict is marginally preferable# to an informative error and much better than a confusing errorreturn{}model_config:ClassVar[SettingsConfigDict]=SettingsConfigDict(extra='forbid',arbitrary_types_allowed=True,validate_default=True,case_sensitive=False,env_prefix='',nested_model_default_partial_update=False,env_file=None,env_file_encoding=None,env_ignore_empty=False,env_nested_delimiter=None,env_nested_max_split=None,env_parse_none_str=None,env_parse_enums=None,cli_prog_name=None,cli_parse_args=None,cli_parse_none_str=None,cli_hide_none_type=False,cli_avoid_json=False,cli_enforce_required=False,cli_use_class_docs_for_groups=False,cli_exit_on_error=True,cli_prefix='',cli_flag_prefix_char='-',cli_implicit_flags=False,cli_ignore_unknown_args=False,cli_kebab_case=False,json_file=None,json_file_encoding=None,yaml_file=None,yaml_file_encoding=None,toml_file=None,secrets_dir=None,protected_namespaces=('model_validate','model_dump','settings_customise_sources'),enable_decoding=True,)classCliApp:""" A utility class for running Pydantic `BaseSettings`, `BaseModel`, or `pydantic.dataclasses.dataclass` as CLI applications. """@staticmethoddef_run_cli_cmd(model:Any,cli_cmd_method_name:str,is_required:bool)->Any:command=getattr(type(model),cli_cmd_method_name,None)ifcommandisNone:ifis_required:raiseSettingsError(f'Error: {type(model).__name__} class is missing {cli_cmd_method_name} entrypoint')returnmodel# If the method is asynchronous, we handle its execution based on the current event loop status.ifinspect.iscoroutinefunction(command):# For asynchronous methods, we have two execution scenarios:# 1. If no event loop is running in the current thread, run the coroutine directly with asyncio.run().# 2. If an event loop is already running in the current thread, run the coroutine in a separate thread to avoid conflicts.try:# Check if an event loop is currently running in this thread.loop=asyncio.get_running_loop()exceptRuntimeError:loop=Noneifloopandloop.is_running():# We're in a context with an active event loop (e.g., Jupyter Notebook).# Running asyncio.run() here would cause conflicts, so we use a separate thread.exception_container=[]defrun_coro()->None:try:# Execute the coroutine in a new event loop in this separate thread.asyncio.run(command(model))exceptExceptionase:exception_container.append(e)thread=threading.Thread(target=run_coro)thread.start()thread.join()ifexception_container:# Propagate exceptions from the separate thread.raiseexception_container[0]else:# No event loop is running; safe to run the coroutine directly.asyncio.run(command(model))else:# For synchronous methods, call them directly.command(model)returnmodel@staticmethoddefrun(model_cls:type[T],cli_args:list[str]|Namespace|SimpleNamespace|dict[str,Any]|None=None,cli_settings_source:CliSettingsSource[Any]|None=None,cli_exit_on_error:bool|None=None,cli_cmd_method_name:str='cli_cmd',**model_init_data:Any,)->T:""" Runs a Pydantic `BaseSettings`, `BaseModel`, or `pydantic.dataclasses.dataclass` as a CLI application. Running a model as a CLI application requires the `cli_cmd` method to be defined in the model class. Args: model_cls: The model class to run as a CLI application. cli_args: The list of CLI arguments to parse. If `cli_settings_source` is specified, this may also be a namespace or dictionary of pre-parsed CLI arguments. Defaults to `sys.argv[1:]`. cli_settings_source: Override the default CLI settings source with a user defined instance. Defaults to `None`. cli_exit_on_error: Determines whether this function exits on error. If model is subclass of `BaseSettings`, defaults to BaseSettings `cli_exit_on_error` value. Otherwise, defaults to `True`. cli_cmd_method_name: The CLI command method name to run. Defaults to "cli_cmd". model_init_data: The model init data. Returns: The ran instance of model. Raises: SettingsError: If model_cls is not subclass of `BaseModel` or `pydantic.dataclasses.dataclass`. SettingsError: If model_cls does not have a `cli_cmd` entrypoint defined. """ifnot(is_pydantic_dataclass(model_cls)oris_model_class(model_cls)):raiseSettingsError(f'Error: {model_cls.__name__} is not subclass of BaseModel or pydantic.dataclasses.dataclass')cli_settings=Nonecli_parse_args=Trueifcli_argsisNoneelsecli_argsifcli_settings_sourceisnotNone:ifisinstance(cli_parse_args,(Namespace,SimpleNamespace,dict)):cli_settings=cli_settings_source(parsed_args=cli_parse_args)else:cli_settings=cli_settings_source(args=cli_parse_args)elifisinstance(cli_parse_args,(Namespace,SimpleNamespace,dict)):raiseSettingsError('Error: `cli_args` must be list[str] or None when `cli_settings_source` is not used')model_init_data['_cli_parse_args']=cli_parse_argsmodel_init_data['_cli_exit_on_error']=cli_exit_on_errormodel_init_data['_cli_settings_source']=cli_settingsifnotissubclass(model_cls,BaseSettings):classCliAppBaseSettings(BaseSettings,model_cls):# type: ignoremodel_config=SettingsConfigDict(nested_model_default_partial_update=True,case_sensitive=True,cli_hide_none_type=True,cli_avoid_json=True,cli_enforce_required=True,cli_implicit_flags=True,cli_kebab_case=True,)model=CliAppBaseSettings(**model_init_data)model_init_data={}forfield_name,field_infoinmodel.model_fields.items():model_init_data[_field_name_for_signature(field_name,field_info)]=getattr(model,field_name)returnCliApp._run_cli_cmd(model_cls(**model_init_data),cli_cmd_method_name,is_required=False)@staticmethoddefrun_subcommand(model:PydanticModel,cli_exit_on_error:bool|None=None,cli_cmd_method_name:str='cli_cmd')->PydanticModel:""" Runs the model subcommand. Running a model subcommand requires the `cli_cmd` method to be defined in the nested model subcommand class. Args: model: The model to run the subcommand from. cli_exit_on_error: Determines whether this function exits with error if no subcommand is found. Defaults to model_config `cli_exit_on_error` value if set. Otherwise, defaults to `True`. cli_cmd_method_name: The CLI command method name to run. Defaults to "cli_cmd". Returns: The ran subcommand model. Raises: SystemExit: When no subcommand is found and cli_exit_on_error=`True` (the default). SettingsError: When no subcommand is found and cli_exit_on_error=`False`. """subcommand=get_subcommand(model,is_required=True,cli_exit_on_error=cli_exit_on_error)returnCliApp._run_cli_cmd(subcommand,cli_cmd_method_name,is_required=True)