Merge branch 'main' into issue-482

Author: hramezani

pydantic_settings/sources.py

@@ -661,7 +661,9 @@ def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str,
                 a flag to determine whether value is complex.
         """
 
-        for field_key, env_name, value_is_complex in self._extract_field_info(field, field_name):
+        field_infos = self._extract_field_info(field, field_name)
+        preferred_key, *_ = field_infos[0]
+        for field_key, env_name, value_is_complex in field_infos:
             # paths reversed to match the last-wins behaviour of `env_file`
             for secrets_path in reversed(self.secrets_paths):
                 path = self.find_case_path(secrets_path, env_name, self.case_sensitive)
@@ -670,14 +672,16 @@ def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str,
                     continue
 
                 if path.is_file():
-                    return path.read_text().strip(), field_key, value_is_complex
+                    if value_is_complex or (self.config.get('populate_by_name', False) and (field_key == field_name)):
+                        preferred_key = field_key
+                    return path.read_text().strip(), preferred_key, value_is_complex
                 else:
                     warnings.warn(
                         f'attempted to load secret file "{path}" but found a {path_type_label(path)} instead.',
                         stacklevel=4,
                     )
 
-        return None, field_key, value_is_complex
+        return None, preferred_key, value_is_complex
 
     def __repr__(self) -> str:
         return f'{self.__class__.__name__}(secrets_dir={self.secrets_dir!r})'
@@ -725,12 +729,16 @@ def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str,
         """
 
         env_val: str | None = None
-        for field_key, env_name, value_is_complex in self._extract_field_info(field, field_name):
+        field_infos = self._extract_field_info(field, field_name)
+        preferred_key, *_ = field_infos[0]
+        for field_key, env_name, value_is_complex in field_infos:
             env_val = self.env_vars.get(env_name)
             if env_val is not None:
+                if value_is_complex or (self.config.get('populate_by_name', False) and (field_key == field_name)):
+                    preferred_key = field_key
                 break
 
-        return env_val, field_key, value_is_complex
+        return env_val, preferred_key, value_is_complex
 
     def prepare_field_value(self, field_name: str, field: FieldInfo, value: Any, value_is_complex: bool) -> Any:
         """
@@ -1426,8 +1434,8 @@ def _get_sub_models(self, model: type[BaseModel], field_name: str, field_info: F
                 raise SettingsError(f'CliSubCommand is not outermost annotation for {model.__name__}.{field_name}')
             elif _annotation_contains_types(type_, (_CliPositionalArg,), is_include_origin=False):
                 raise SettingsError(f'CliPositionalArg is not outermost annotation for {model.__name__}.{field_name}')
-            if is_model_class(type_) or is_pydantic_dataclass(type_):
-                sub_models.append(type_)  # type: ignore
+            if is_model_class(_strip_annotated(type_)) or is_pydantic_dataclass(_strip_annotated(type_)):
+                sub_models.append(_strip_annotated(type_))
         return sub_models
 
     def _verify_cli_flag_annotations(self, model: type[BaseModel], field_name: str, field_info: FieldInfo) -> None:

tests/test_source_cli.py

@@ -10,11 +10,14 @@
 import typing_extensions
 from pydantic import (
     AliasChoices,
+    AliasGenerator,
     AliasPath,
     BaseModel,
     ConfigDict,
     DirectoryPath,
+    Discriminator,
     Field,
+    Tag,
     ValidationError,
 )
 from pydantic import (
@@ -107,7 +110,7 @@ def parse_args(self, *args: Any, **kwargs: Any) -> argparse.Namespace:
         return self.parser.parse_args(*args, **kwargs)
 
 
-def test_validation_alias_with_cli_prefix():
+def test_cli_validation_alias_with_cli_prefix():
     class Settings(BaseSettings, cli_exit_on_error=False):
         foobar: str = Field(validation_alias='foo')
 
@@ -119,6 +122,36 @@ class Settings(BaseSettings, cli_exit_on_error=False):
     assert CliApp.run(Settings, cli_args=['--p.foo', 'bar']).foobar == 'bar'
 
 
+@pytest.mark.parametrize(
+    'alias_generator',
+    [
+        AliasGenerator(validation_alias=lambda s: AliasChoices(s, s.replace('_', '-'))),
+        AliasGenerator(validation_alias=lambda s: AliasChoices(s.replace('_', '-'), s)),
+    ],
+)
+def test_cli_alias_resolution_consistency_with_env(env, alias_generator):
+    class SubModel(BaseModel):
+        v1: str = 'model default'
+
+    class Settings(BaseSettings):
+        model_config = SettingsConfigDict(
+            env_nested_delimiter='__',
+            nested_model_default_partial_update=True,
+            alias_generator=alias_generator,
+        )
+
+        sub_model: SubModel = SubModel(v1='top default')
+
+    assert CliApp.run(Settings, cli_args=[]).model_dump() == {'sub_model': {'v1': 'top default'}}
+
+    env.set('SUB_MODEL__V1', 'env default')
+    assert CliApp.run(Settings, cli_args=[]).model_dump() == {'sub_model': {'v1': 'env default'}}
+
+    assert CliApp.run(Settings, cli_args=['--sub-model.v1=cli default']).model_dump() == {
+        'sub_model': {'v1': 'cli default'}
+    }
+
+
 def test_cli_nested_arg():
     class SubSubValue(BaseModel):
         v6: str
@@ -2237,3 +2270,25 @@ class MySettings(BaseSettings):
         CliApp.run(
             MySettings, cli_args=['--bac', 'cli abbrev are invalid for internal parser'], cli_exit_on_error=False
         )
+
+
+def test_cli_submodels_strip_annotated():
+    class PolyA(BaseModel):
+        a: int = 1
+        type: Literal['a'] = 'a'
+
+    class PolyB(BaseModel):
+        b: str = '2'
+        type: Literal['b'] = 'b'
+
+    def _get_type(model: Union[BaseModel, Dict]) -> str:
+        if isinstance(model, dict):
+            return model.get('type', 'a')
+        return model.type  # type: ignore
+
+    Poly = Annotated[Union[Annotated[PolyA, Tag('a')], Annotated[PolyB, Tag('b')]], Discriminator(_get_type)]
+
+    class WithUnion(BaseSettings):
+        poly: Poly
+
+    assert CliApp.run(WithUnion, ['--poly.type=a']).model_dump() == {'poly': {'a': 1, 'type': 'a'}}