Fix nested discriminated unions not discovered by env/CLI providers (#816)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: hramezani

pydantic_settings/sources/providers/cli.py

@@ -90,6 +90,16 @@ class CliMutuallyExclusiveGroup(BaseModel):
     pass
 
 
+def _collect_sub_models(type_: Any, sub_models: list[type[BaseModel]]) -> None:
+    """Recursively collect BaseModel subclasses from possibly nested union types."""
+    stripped = _strip_annotated(type_)
+    if is_model_class(stripped) or is_pydantic_dataclass(stripped):
+        sub_models.append(stripped)  # type: ignore[arg-type]
+    elif is_union_origin(get_origin(stripped)):
+        for arg in get_args(stripped):
+            _collect_sub_models(arg, sub_models)
+
+
 class _CliArg(BaseModel):
     model: Any
     parser: Any
@@ -200,8 +210,7 @@ def sub_models(self) -> list[type[BaseModel]]:
                 raise SettingsError(
                     f'CliPositionalArg is not outermost annotation for {self.model.__name__}.{self.field_name}'
                 )
-            if is_model_class(_strip_annotated(type_)) or is_pydantic_dataclass(_strip_annotated(type_)):
-                sub_models.append(_strip_annotated(type_))
+            _collect_sub_models(type_, sub_models)
         return sub_models
 
     @cached_property

pydantic_settings/sources/utils.py

@@ -134,7 +134,22 @@ def _annotation_is_complex_inner(annotation: type[Any] | None) -> bool:
 
 def _union_is_complex(annotation: type[Any] | None, metadata: list[Any]) -> bool:
     """Check if a union type contains any complex types."""
-    return any(_annotation_is_complex(arg, metadata) for arg in get_args(annotation))
+    for arg in get_args(annotation):
+        if _annotation_is_complex(arg, metadata):
+            return True
+        # _annotation_is_complex doesn't handle bare Union types, so when an arg
+        # is Annotated[Union[X, Y], ...], stripping Annotated yields a bare Union
+        # that _annotation_is_complex can't evaluate.  Recurse into it, but only
+        # if the Annotated metadata doesn't suppress complexity (e.g. Json).
+        inner = _strip_annotated(arg)
+        if inner is not arg:
+            _, *inner_meta = get_args(arg)
+            if any(isinstance(md, Json) for md in inner_meta):  # type: ignore[misc]
+                continue
+        if is_union_origin(get_origin(inner)):
+            if _union_is_complex(inner, metadata):
+                return True
+    return False
 
 
 def _union_has_strict_types(annotation: type[Any] | None) -> bool:

tests/test_settings.py

@@ -2362,6 +2362,31 @@ class Settings(BaseSettings):
     assert s.a_or_b.x == 'a'
 
 
+def test_nested_discriminated_union(env):
+    """Test that nested unions like Union[Annotated[Union[A, B], Discriminator(...)], None] are handled correctly."""
+
+    class A(BaseModel):
+        x: Literal['a'] = 'a'
+        a: int = 1
+
+    class B(BaseModel):
+        x: Literal['b'] = 'b'
+        b: str = 'hello'
+
+    AOrB = Annotated[A | B, Discriminator('x')]
+
+    class Settings(BaseSettings):
+        model_config = SettingsConfigDict(env_nested_delimiter='__')
+        a_or_b: AOrB | None = None
+
+    # Test env var for nested model field
+    env.set('a_or_b__x', 'b')
+    env.set('a_or_b__b', 'world')
+    s = Settings()
+    assert isinstance(s.a_or_b, B)
+    assert s.a_or_b.b == 'world'
+
+
 def test_nested_model_case_insensitive(env):
     class SubSubSub(BaseModel):
         VaL3: str

tests/test_source_cli.py

@@ -1449,6 +1449,31 @@ class Cfg(BaseSettings):
     assert cfg.model_dump() == {'child': {'name': 'new name a', 'diff_a': 'new diff a'}}
 
 
+def test_cli_nested_discriminated_union():
+    """Test that nested unions like Union[Annotated[Union[A, B], Discriminator(...)], None] are handled by CLI."""
+
+    class A(BaseModel):
+        x: Literal['a'] = 'a'
+        a: int = 1
+
+    class B(BaseModel):
+        x: Literal['b'] = 'b'
+        b: str = 'hello'
+
+    AOrB = Annotated[A | B, Discriminator('x')]
+
+    class Cfg(BaseSettings):
+        a_or_b: AOrB | None = None
+
+    cfg = CliApp.run(Cfg, cli_args=['--a_or_b.x', 'b', '--a_or_b.b', 'world'])
+    assert isinstance(cfg.a_or_b, B)
+    assert cfg.a_or_b.b == 'world'
+
+    cfg = CliApp.run(Cfg, cli_args=['--a_or_b.x', 'a', '--a_or_b.a', '42'])
+    assert isinstance(cfg.a_or_b, A)
+    assert cfg.a_or_b.a == 42
+
+
 def test_cli_optional_positional_arg(env):
     class Main(BaseSettings):
         model_config = SettingsConfigDict(