Add CLI Settings Source

docs/index.md

@@ -363,7 +363,6 @@ class Settings(BaseSettings):
         cls,
         settings_cls: Type[BaseSettings],
         init_settings: PydanticBaseSettingsSource,
-        cli_settings: PydanticBaseSettingsSource,
         env_settings: PydanticBaseSettingsSource,
         dotenv_settings: PydanticBaseSettingsSource,
         file_secret_settings: PydanticBaseSettingsSource,
@@ -576,7 +575,6 @@ class Settings(BaseSettings):
         cls,
         settings_cls: Type[BaseSettings],
         init_settings: PydanticBaseSettingsSource,
-        cli_settings: PydanticBaseSettingsSource,
         env_settings: PydanticBaseSettingsSource,
         dotenv_settings: PydanticBaseSettingsSource,
         file_secret_settings: PydanticBaseSettingsSource,
@@ -659,7 +657,6 @@ class Settings(BaseSettings):
         cls,
         settings_cls: Type[BaseSettings],
         init_settings: PydanticBaseSettingsSource,
-        cli_settings: PydanticBaseSettingsSource,
         env_settings: PydanticBaseSettingsSource,
         dotenv_settings: PydanticBaseSettingsSource,
         file_secret_settings: PydanticBaseSettingsSource,
@@ -696,7 +693,6 @@ class Settings(BaseSettings):
         cls,
         settings_cls: Type[BaseSettings],
         init_settings: PydanticBaseSettingsSource,
-        cli_settings: PydanticBaseSettingsSource,
         env_settings: PydanticBaseSettingsSource,
         dotenv_settings: PydanticBaseSettingsSource,
         file_secret_settings: PydanticBaseSettingsSource,

pydantic_settings/main.py

@@ -32,6 +32,7 @@ class SettingsConfigDict(ConfigDict, total=False):
     cli_parse_args: bool | list[str] | None
     cli_hide_none_type: bool
     cli_avoid_json: bool
+    cli_enforce_required: bool
     secrets_dir: str | Path | None
 
 
@@ -70,6 +71,7 @@ class BaseSettings(BaseModel):
             If set to `True`, defaults to sys.argv[1:].
         _cli_hide_none_type: Hide NoneType 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`.
         _secrets_dir: The secret files directory. Defaults to `None`.
     """
 
@@ -86,6 +88,7 @@ def __init__(
         _cli_parse_args: bool | list[str] | None = None,
         _cli_hide_none_type: bool | None = None,
         _cli_avoid_json: bool | None = None,
+        _cli_enforce_required: bool | None = None,
         _secrets_dir: str | Path | None = None,
         **values: Any,
     ) -> None:
@@ -104,6 +107,7 @@ def __init__(
                 _cli_parse_args=_cli_parse_args,
                 _cli_hide_none_type=_cli_hide_none_type,
                 _cli_avoid_json=_cli_avoid_json,
+                _cli_enforce_required=_cli_enforce_required,
                 _secrets_dir=_secrets_dir,
             )
         )
@@ -113,7 +117,6 @@ def settings_customise_sources(
         cls,
         settings_cls: type[BaseSettings],
         init_settings: PydanticBaseSettingsSource,
-        cli_settings: PydanticBaseSettingsSource,
         env_settings: PydanticBaseSettingsSource,
         dotenv_settings: PydanticBaseSettingsSource,
         file_secret_settings: PydanticBaseSettingsSource,
@@ -124,15 +127,14 @@ def settings_customise_sources(
         Args:
             settings_cls: The Settings class.
             init_settings: The `InitSettingsSource` instance.
-            cli_settings: The `CliSettingsSource` 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.
         """
-        return init_settings, cli_settings, env_settings, dotenv_settings, file_secret_settings
+        return init_settings, env_settings, dotenv_settings, file_secret_settings
 
     def _settings_build_values(
         self,
@@ -148,6 +150,7 @@ def _settings_build_values(
         _cli_parse_args: bool | list[str] | None = None,
         _cli_hide_none_type: bool | None = None,
         _cli_avoid_json: bool | None = None,
+        _cli_enforce_required: bool | None = None,
         _secrets_dir: str | Path | None = None,
     ) -> dict[str, Any]:
         # Determine settings config values
@@ -175,6 +178,11 @@ def _settings_build_values(
             _cli_hide_none_type if _cli_hide_none_type is not None else self.model_config.get('cli_hide_none_type')
         )
         cli_avoid_json = _cli_avoid_json if _cli_avoid_json is not None else self.model_config.get('cli_avoid_json')
+        cli_enforce_required = (
+            _cli_enforce_required
+            if _cli_enforce_required is not None
+            else self.model_config.get('cli_enforce_required')
+        )
 
         secrets_dir = _secrets_dir if _secrets_dir is not None else self.model_config.get('secrets_dir')
 
@@ -187,6 +195,7 @@ def _settings_build_values(
             cli_parse_none_str=env_parse_none_str,
             cli_hide_none_type=cli_hide_none_type,
             cli_avoid_json=cli_avoid_json,
+            cli_enforce_required=cli_enforce_required,
         )
         env_settings = EnvSettingsSource(
             self.__class__,
@@ -214,11 +223,12 @@ def _settings_build_values(
         sources = self.settings_customise_sources(
             self.__class__,
             init_settings=init_settings,
-            cli_settings=cli_settings,
             env_settings=env_settings,
             dotenv_settings=dotenv_settings,
             file_secret_settings=file_secret_settings,
         )
+        if _cli_parse_args:
+            sources = (cli_settings,) + sources
         if sources:
             return deep_update(*reversed([source() for source in sources]))
         else:
@@ -241,6 +251,7 @@ def _settings_build_values(
         cli_parse_args=None,
         cli_hide_none_type=False,
         cli_avoid_json=False,
+        cli_enforce_required=False,
         secrets_dir=None,
         protected_namespaces=('model_', 'settings_'),
     )

pydantic_settings/sources.py

@@ -7,7 +7,6 @@
 from abc import ABC, abstractmethod
 from collections import deque
 from dataclasses import is_dataclass
-from inspect import isclass
 from pathlib import Path
 from types import FunctionType
 from typing import TYPE_CHECKING, Any, Dict, List, Literal, Mapping, Sequence, Tuple, TypeVar, Union, cast
@@ -16,7 +15,7 @@
 from pydantic import AliasChoices, AliasPath, BaseModel, Json, TypeAdapter
 from pydantic._internal._repr import Representation
 from pydantic._internal._typing_extra import WithArgsTypes, origin_is_union, typing_base
-from pydantic._internal._utils import deep_update, lenient_issubclass
+from pydantic._internal._utils import deep_update, is_model_class, lenient_issubclass
 from pydantic.fields import FieldInfo
 from typing_extensions import Annotated, TypeAliasType, get_args, get_origin
 
@@ -515,7 +514,7 @@
         return True, allow_parse_failure
 
     @staticmethod
-    def next_field(field: FieldInfo | None, key: str) -> FieldInfo | None:
+    def next_field(field: FieldInfo | Any | None, key: str) -> FieldInfo | None:
         """
         Find the field in a sub model by key(env name)
 
@@ -544,11 +543,25 @@
         Returns:
             Field if it finds the next field otherwise `None`.
         """
-        if not field or origin_is_union(get_origin(field.annotation)):
-            # no support for Unions of complex BaseSettings fields
+        if not field:
             return None
-        elif field.annotation and hasattr(field.annotation, 'model_fields') and field.annotation.model_fields.get(key):
-            return field.annotation.model_fields[key]
+        if isinstance(field, FieldInfo):
+            if not hasattr(field, 'annotation'):
+                return None
+            annotation = field.annotation
+        else:
+            annotation = field
+
+        if origin_is_union(get_origin(annotation)) or isinstance(annotation, WithArgsTypes):
+            type_ = get_origin(annotation)
+            if is_model_class(type_) and type_.model_fields.get(key):
+                return type_.model_fields.get(key)
+            for type_ in get_args(annotation):
+                type_has_key = EnvSettingsSource.next_field(type_, key)
+                if type_has_key:
+                    return type_has_key
+        elif is_model_class(annotation) and annotation.model_fields.get(key):
+            return annotation.model_fields[key]
 
         return None
 
@@ -697,20 +710,24 @@
         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,
     ) -> None:
         self.cli_prog_name = sys.argv[0] if cli_prog_name is None else cli_prog_name
         self.cli_parse_args = cli_parse_args
         if self.cli_parse_args not in (None, False):
             if self.cli_parse_args is True:
                 self.cli_parse_args = sys.argv[1:]
             elif not isinstance(self.cli_parse_args, list):
                 raise SettingsError(f'cli_parse_args must be List[str], recieved {type(self.cli_parse_args)}')
         self.cli_hide_none_type = (
             cli_hide_none_type if cli_hide_none_type is not None else self.config.get('cli_hide_none_type', False)
         )
         self.cli_avoid_json = cli_avoid_json if cli_avoid_json is not None else self.config.get('cli_avoid_json', False)
         if cli_parse_none_str is None:
             cli_parse_none_str = 'None' if self.cli_avoid_json is True else 'null'
+        self.cli_enforce_required = (
+            cli_enforce_required if cli_enforce_required is not None else self.config.get('cli_enforce_required', False)
+        )
         super().__init__(settings_cls, env_nested_delimiter='.', env_parse_none_str=cli_parse_none_str)
 
     def _load_env_vars(self) -> Mapping[str, str | None]:
@@ -729,25 +746,32 @@
         )
 
         parsed_args: dict[str, list[str] | str] = vars(parser.parse_args(self.cli_parse_args))  # type: ignore
-        if any(key for key in parsed_args.keys() if not key.endswith(':subcommand')):
-            for field, val in parsed_args.items():
-                if isinstance(val, list):
-                    merge_list = []
-                    for sub_val in val:
-                        if sub_val.startswith('[') and sub_val.endswith(']'):
-                            sub_val = sub_val[1:-1]
-                        merge_list.append(sub_val)
-                    parsed_args[field] = (
-                        f'[{",".join(merge_list)}]'
-                        if field not in self._cli_dict_arg_names
-                        else self._merge_json_key_val_list_str(f'[{",".join(merge_list)}]')
-                    )
-                elif field.endswith(':subcommand'):
-                    self._cli_subcommands[field].remove(field.split(':')[0] + val)
+        selected_subcommands: list[str] = []
+        for field_name, val in parsed_args.items():
+            if isinstance(val, list):
+                merge_list = []
+                for sub_val in val:
+                    if sub_val.startswith('[') and sub_val.endswith(']'):
+                        sub_val = sub_val[1:-1]
+                    merge_list.append(sub_val)
+                parsed_args[field_name] = (
+                    f'[{",".join(merge_list)}]'
+                    if field_name not in self._cli_dict_arg_names
+                    else self._merge_json_key_val_list_str(f'[{",".join(merge_list)}]')
+                )
+            elif field_name.endswith(':subcommand'):
+                selected_subcommands.append(field_name.split(':')[0] + val)
 
         for subcommands in self._cli_subcommands.values():
             for subcommand in subcommands:
-                parsed_args[subcommand] = self.env_parse_none_str  # type: ignore
+                if subcommand not in selected_subcommands:
+                    parsed_args[subcommand] = self.env_parse_none_str  # type: ignore
+
+        parsed_args = {key: val for key, val in parsed_args.items() if not key.endswith(':subcommand')}
+        if selected_subcommands:
+            last_selected_subcommand = max(selected_subcommands, key=len)
+            if not any(field_name for field_name in parsed_args.keys() if f'{last_selected_subcommand}.' in field_name):
+                parsed_args[last_selected_subcommand] = '{}'
 
         return parse_env_vars(
             parsed_args, self.case_sensitive, self.env_ignore_empty, self.env_parse_none_str  # type: ignore
@@ -759,7 +783,7 @@
         obj_count = 0
         while key_val_list_str:
             if obj_count != 0:
                 raise SettingsError(f'Parsing error encountered on JSON object {orig_key_val_list_str}')
             for i in range(len(key_val_list_str)):
                 if key_val_list_str[i] == '{':
                     obj_count += 1
@@ -800,7 +824,7 @@
                     raise SettingsError(
                         f'CliPositionalArg is not outermost annotation for {model.__name__}.{field_name}'
                     )
-            if isclass(type_) and issubclass(type_, BaseModel):
+            if is_model_class(type_):
                 sub_models.append(type_)
         return sub_models
 
@@ -814,7 +838,7 @@
                     field_types = [type_ for type_ in get_args(field_info.annotation) if type_ is not type(None)]
                     if len(field_types) != 1:
                         raise SettingsError(f'subcommand argument {model.__name__}.{field_name} has multiple types')
-                    elif not (isclass(field_types[0]) and issubclass(field_types[0], BaseModel)):
+                    elif not is_model_class(field_types[0]):
                         raise SettingsError(
                             f'subcommand argument {model.__name__}.{field_name} is not derived from BaseModel'
                         )
@@ -867,6 +891,7 @@
                 kwargs['help'] = field_info.description
                 kwargs['dest'] = f'{arg_prefix}{field_name}'
                 kwargs['metavar'] = self._format_metavar(field_info.annotation)
+                kwargs['required'] = self.cli_enforce_required and field_info.is_required()
                 if kwargs['dest'] in added_args:
                     continue
                 if _annotation_contains_types(field_info.annotation, (list, set, dict, Sequence, Mapping)):
@@ -879,6 +904,7 @@
                     kwargs['metavar'] = field_name.upper()
                     arg_name = kwargs['dest']
                     del kwargs['dest']
+                    del kwargs['required']
                     arg_flag = ''
 
                 if sub_models and kwargs.get('action') != 'append':
@@ -913,36 +939,36 @@
     def _format_metavar(self, obj: Any) -> str:
         """Pretty metavar representation of a type. Adapts logic from `pydantic._repr.display_as_type`."""
         if isinstance(obj, FunctionType):
             return obj.__name__
         elif obj is ...:
             return '...'
         elif isinstance(obj, Representation):
             return repr(obj)
         elif isinstance(obj, TypeAliasType):
             return str(obj)
 
         if not isinstance(obj, (typing_base, WithArgsTypes, type)):
             obj = obj.__class__
 
         if origin_is_union(get_origin(obj)):
             args = ','.join(map(self._format_metavar, self._get_modified_args(obj)))
             return f'{{{args}}}' if ',' in args else args
         elif isinstance(obj, WithArgsTypes):
             if get_origin(obj) == Literal:
                 args = ','.join(map(repr, self._get_modified_args(obj)))
                 return f'{{{args}}}' if ',' in args else args
             else:
                 args = ','.join(map(self._format_metavar, self._get_modified_args(obj)))
             try:
                 return f'{obj.__qualname__}[{args}]'
             except AttributeError:
                 return str(obj)  # handles TypeAliasType in 3.12
         elif obj is type(None):
             return self.env_parse_none_str
         elif isinstance(obj, type):
             return obj.__qualname__
         else:
             return repr(obj).replace('typing.', '').replace('typing_extensions.', '')
 
 
 def _get_env_var_key(key: str, case_sensitive: bool = False) -> str: