Fix an issue when inner types of a discriminated union with a callable discriminator were not correctly identified as complex. (#285)

Author: jenskeiner

pydantic_settings/sources.py

@@ -16,7 +16,7 @@
 from pydantic._internal._typing_extra import WithArgsTypes, origin_is_union
 from pydantic._internal._utils import deep_update, is_model_class, lenient_issubclass
 from pydantic.fields import FieldInfo
-from typing_extensions import get_args, get_origin
+from typing_extensions import _AnnotatedAlias, get_args, get_origin
 
 from pydantic_settings.utils import path_type_label
 
@@ -913,6 +913,11 @@ def read_env_file(
 def _annotation_is_complex(annotation: type[Any] | None, metadata: list[Any]) -> bool:
     if any(isinstance(md, Json) for md in metadata):  # type: ignore[misc]
         return False
+    # Check if annotation is of the form Annotated[type, metadata].
+    if isinstance(annotation, _AnnotatedAlias):
+        # Return result of recursive call on inner type.
+        inner, meta = get_args(annotation)
+        return _annotation_is_complex(inner, [meta])
     origin = get_origin(annotation)
     return (
         _annotation_is_complex_inner(annotation)

tests/test_settings.py

@@ -6,27 +6,29 @@
 from datetime import datetime, timezone
 from enum import IntEnum
 from pathlib import Path
-from typing import Any, Callable, Dict, Generic, List, Optional, Set, Tuple, Type, TypeVar, Union
+from typing import Any, Callable, Dict, Generic, Hashable, List, Optional, Set, Tuple, Type, TypeVar, Union
 
 import pytest
 from annotated_types import MinLen
 from pydantic import (
     AliasChoices,
     AliasPath,
     BaseModel,
+    Discriminator,
     Field,
     HttpUrl,
     Json,
     RootModel,
     SecretStr,
+    Tag,
     ValidationError,
 )
 from pydantic import (
     dataclasses as pydantic_dataclasses,
 )
 from pydantic.fields import FieldInfo
 from pytest_mock import MockerFixture
-from typing_extensions import Annotated
+from typing_extensions import Annotated, Literal
 
 from pydantic_settings import (
     BaseSettings,
@@ -1674,6 +1676,44 @@ class Cfg(BaseSettings):
         Cfg()
 
 
+def test_discriminated_union_with_callable_discriminator(env):
+    class A(BaseModel):
+        x: Literal['a'] = 'a'
+        y: str
+
+    class B(BaseModel):
+        x: Literal['b'] = 'b'
+        z: str
+
+    def get_discriminator_value(v: Any) -> Hashable:
+        if isinstance(v, dict):
+            v0 = v.get('x')
+        else:
+            v0 = getattr(v, 'x', None)
+
+        if v0 == 'a':
+            return 'a'
+        elif v0 == 'b':
+            return 'b'
+        else:
+            return None
+
+    class Settings(BaseSettings):
+        model_config = SettingsConfigDict(env_nested_delimiter='__')
+
+        # Discriminated union using a callable discriminator.
+        a_or_b: Annotated[Union[Annotated[A, Tag('a')], Annotated[B, Tag('b')]], Discriminator(get_discriminator_value)]
+
+    # Set up environment so that the discriminator is 'a'.
+    env.set('a_or_b__x', 'a')
+    env.set('a_or_b__y', 'foo')
+
+    s = Settings()
+
+    assert s.a_or_b.x == 'a'
+    assert s.a_or_b.y == 'foo'
+
+
 def test_nested_model_case_insensitive(env):
     class SubSubSub(BaseModel):
         VaL3: str