Check union args to don't consider Optional fields as complex Union

pydantic_settings/sources.py

@@ -13,7 +13,7 @@
 from pydantic._internal._typing_extra import origin_is_union
 from pydantic._internal._utils import deep_update, lenient_issubclass
 from pydantic.fields import FieldInfo
-from typing_extensions import get_origin
+from typing_extensions import get_args, get_origin
 
 from pydantic_settings.utils import path_type_label
 
@@ -441,13 +441,22 @@ def prepare_field_value(self, field_name: str, field: FieldInfo, value: Any, val
             # simplest case, field is not complex, we only need to add the value if it was found
             return value
 
+    def _union_is_complex(self, annotation: type[Any] | None, metadata: list[Any]) -> bool:
+        for arg in get_args(annotation):
+            if arg is type(None):  # Optional
+                continue
+            elif _annotation_is_complex(arg, metadata):
+                return True
+
+        return False
+
     def _field_is_complex(self, field: FieldInfo) -> tuple[bool, bool]:
         """
         Find out if a field is complex, and if so whether JSON errors should be ignored
         """
         if self.field_is_complex(field):
             allow_parse_failure = False
-        elif origin_is_union(get_origin(field.annotation)):
+        elif origin_is_union(get_origin(field.annotation)) and self._union_is_complex(field.annotation, field.metadata):
             allow_parse_failure = True
         else:
             return False, False

tests/test_settings.py

@@ -1791,3 +1791,28 @@ class Settings(BaseSettings):
     env.set('z', '[{"x": 1, "y": {"foo": 1}}, {"x": 2, "y": {"foo": 2}}]')
     s = Settings()
     assert s.model_dump() == {'z': [{'x': 1, 'y': {'foo': 1}}, {'x': 2, 'y': {'foo': 2}}]}
+
+
+def test_optional_field_from_env(env):
+    class Settings(BaseSettings):
+        x: Optional[str] = None
+
+    env.set('x', '123')
+
+    s = Settings()
+    assert s.x == '123'
+
+
+@pytest.mark.skipif(not dotenv, reason='python-dotenv not installed')
+def test_dotenv_optional_json_field(tmp_path):
+    p = tmp_path / '.env'
+    p.write_text("""DATA='{"foo":"bar"}'""")
+
+    class Settings(BaseSettings):
+        model_config = SettingsConfigDict(env_file=p)
+
+        data: Optional[Json[dict[str, str]]] = Field(default=None)
+
+    s = Settings()
+    print(s.data)
+    assert s.data == {'foo': 'bar'}