Implement raise_on_missing_reader flag for LoadImage transforms

This commit addresses CodeRabbit review feedback and adds a new flag to control
exception behavior when readers are not available:

- Add raise_on_missing_reader parameter to LoadImage and LoadImaged
- Fix guarding for class-based readers to respect the flag
- Add proper stacklevel and category to all warning messages for better UX
- Update LoadImage docstring with Raises section and accepted reader types
- Add comprehensive test coverage for the new functionality
- Apply code formatting with black, isort, and ruff
- Maintain backward compatibility (flag defaults to False)

The flag allows users to choose between strict error handling or fallback
behavior when specified readers cannot be found or their dependencies
are missing.

Signed-off-by: Sarah <[email protected]>

Author: its-serah

monai/transforms/io/array.py

@@ -45,7 +45,10 @@
 from monai.transforms.utility.array import EnsureChannelFirst
 from monai.utils import (
     GridSamplePadMode,
-    ImageMetaKey,
+)
+from monai.utils import ImageMetaKey
+from monai.utils import ImageMetaKey as Key
+from monai.utils import (
     MetaKeys,
     OptionalImportError,
     convert_to_dst_type,
@@ -138,6 +141,7 @@ def __init__(
         prune_meta_pattern: str | None = None,
         prune_meta_sep: str = ".",
         expanduser: bool = True,
+        raise_on_missing_reader: bool = False,
         *args,
         **kwargs,
     ) -> None:
@@ -161,9 +165,21 @@ def __init__(
                 in the metadata (nested dictionary). default is ".", see also :py:class:`monai.transforms.DeleteItemsd`.
                 e.g. ``prune_meta_pattern=".*_code$", prune_meta_sep=" "`` removes meta keys that ends with ``"_code"``.
             expanduser: if True cast filename to Path and call .expanduser on it, otherwise keep filename as is.
+            raise_on_missing_reader: if True, raise `OptionalImportError` when a specified reader is not available;
+                otherwise attempt to use fallback readers. Defaults to False (backward compatibility).
             args: additional parameters for reader if providing a reader name.
             kwargs: additional parameters for reader if providing a reader name.
 
+        Raises:
+            OptionalImportError: If `raise_on_missing_reader=True` and the specified reader
+                cannot be found or its optional dependency is not installed.
+
+        Accepted reader types:
+            - str: name of a registered reader (e.g., `"ITKReader"`)
+            - class: e.g., `ITKReader` or a custom reader class
+            - instance: e.g., `ITKReader(pixel_type=itk.UC)`
+            - list/tuple: multiple reader names or classes to try in order
+
         Note:
 
             - The transform returns a MetaTensor, unless `set_track_meta(False)` has been used, in which case, a
@@ -183,6 +199,7 @@ def __init__(
         self.pattern = prune_meta_pattern
         self.sep = prune_meta_sep
         self.expanduser = expanduser
+        self.raise_on_missing_reader = raise_on_missing_reader
 
         self.readers: list[ImageReader] = []
         for r in SUPPORTED_READERS:  # set predefined readers as default
@@ -206,18 +223,61 @@ def __init__(
                 if not has_built_in:
                     the_reader = locate(f"{_r}")  # search dotted path
                 if the_reader is None:
-                    the_reader = look_up_option(_r.lower(), SUPPORTED_READERS)
+                    try:
+                        the_reader = look_up_option(_r.lower(), SUPPORTED_READERS)
+                    except ValueError:
+                        # If the reader name is not recognized at all, raise OptionalImportError
+                        msg = f"Cannot find reader '{_r}'. It may not be installed or recognized."
+                        if self.raise_on_missing_reader:
+                            raise OptionalImportError(msg)
+                        else:
+                            warnings.warn(
+                                f"{msg} Will use fallback readers if available.",
+                                category=UserWarning,
+                                stacklevel=2,
+                            )
+                            continue
                 try:
                     self.register(the_reader(*args, **kwargs))
-                except OptionalImportError:
-                    warnings.warn(
-                        f"required package for reader {_r} is not installed, or the version doesn't match requirement."
+                except OptionalImportError as e:
+                    msg = (
+                        f"Required package for reader {_r} is not installed, or the version doesn't match requirement."
                     )
+                    if self.raise_on_missing_reader:
+                        raise OptionalImportError(msg) from e
+                    else:
+                        warnings.warn(
+                            f"{msg} Will use fallback readers if available.",
+                            category=UserWarning,
+                            stacklevel=2,
+                        )
                 except TypeError:  # the reader doesn't have the corresponding args/kwargs
-                    warnings.warn(f"{_r} is not supported with the given parameters {args} {kwargs}.")
+                    warnings.warn(
+                        f"{_r} is not supported with the given parameters {args} {kwargs}.",
+                        category=UserWarning,
+                        stacklevel=2,
+                    )
                     self.register(the_reader())
             elif inspect.isclass(_r):
-                self.register(_r(*args, **kwargs))
+                try:
+                    self.register(_r(*args, **kwargs))
+                except OptionalImportError as e:
+                    msg = f"Required package for reader {_r.__name__} is not installed, or the version doesn't match requirement."
+                    if self.raise_on_missing_reader:
+                        raise OptionalImportError(msg) from e
+                    else:
+                        warnings.warn(
+                            f"{msg} Will use fallback readers if available.",
+                            category=UserWarning,
+                            stacklevel=2,
+                        )
+                except TypeError:
+                    warnings.warn(
+                        f"{_r.__name__} is not supported with the given parameters {args} {kwargs}.",
+                        category=UserWarning,
+                        stacklevel=2,
+                    )
+                    self.register(_r())
             else:
                 self.register(_r)  # reader instance, ignoring the constructor args/kwargs
         return

tests/transforms/test_load_image.py

@@ -15,6 +15,7 @@
 import shutil
 import tempfile
 import unittest
+import warnings
 from pathlib import Path
 
 import nibabel as nib
@@ -28,7 +29,7 @@
 from monai.data.meta_obj import set_track_meta
 from monai.data.meta_tensor import MetaTensor
 from monai.transforms import LoadImage
-from monai.utils import optional_import
+from monai.utils import OptionalImportError, optional_import
 from tests.test_utils import SkipIfNoModule, assert_allclose, skip_if_downloading_fails, testing_data_config
 
 itk, has_itk = optional_import("itk", allow_namespace_pkg=True)
@@ -436,12 +437,40 @@ def test_my_reader(self):
         self.assertEqual(out.meta["name"], "my test")
         out = LoadImage(image_only=True, reader=_MiniReader, is_compatible=False)("test")
         self.assertEqual(out.meta["name"], "my test")
+
+    def test_reader_not_installed_exception(self):
+        """test if an exception is raised when a specified reader is not installed"""
+        with self.assertRaises(OptionalImportError):
+            LoadImage(image_only=True, reader="NonExistentReader")("test")
         for item in (_MiniReader, _MiniReader(is_compatible=False)):
             out = LoadImage(image_only=True, reader=item)("test")
             self.assertEqual(out.meta["name"], "my test")
         out = LoadImage(image_only=True)("test", reader=_MiniReader(is_compatible=False))
         self.assertEqual(out.meta["name"], "my test")
 
+    def test_raise_on_missing_reader_flag(self):
+        """test raise_on_missing_reader flag behavior"""
+        # Test with flag enabled - should raise exception for unknown reader name
+        with self.assertRaises(OptionalImportError):
+            LoadImage(image_only=True, reader="UnknownReaderName", raise_on_missing_reader=True)
+
+        # Test with flag disabled - should warn but not raise exception for unknown reader name
+        # This should succeed and create the loader with fallback behavior
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            loader_with_fallback = LoadImage(image_only=True, reader="UnknownReaderName", raise_on_missing_reader=False)
+            self.assertIsInstance(loader_with_fallback, LoadImage)
+            # Should have produced a warning about the unknown reader
+            self.assertTrue(any("Cannot find reader 'UnknownReaderName'" in str(warning.message) for warning in w))
+
+        # The flag should work properly with valid readers too
+        loader_with_flag = LoadImage(image_only=True, reader="ITKReader", raise_on_missing_reader=False)
+        loader_without_flag = LoadImage(image_only=True, reader="ITKReader")
+
+        # Both should work since ITK is available in this test environment
+        self.assertIsInstance(loader_with_flag, LoadImage)
+        self.assertIsInstance(loader_without_flag, LoadImage)
+
     def test_itk_meta(self):
         """test metadata from a directory"""
         out = LoadImage(image_only=True, reader="ITKReader", pixel_type=itk_uc, series_meta=True)(