fix

fugue_ray/_utils/io.py

@@ -7,7 +7,7 @@
 from packaging import version
 from pyarrow import csv as pacsv
 from pyarrow import json as pajson
-from ray.data.datasource import FileExtensionFilter
+
 from triad.collections import Schema
 from triad.collections.dict import ParamDict
 from triad.utils.assertion import assert_or_throw
@@ -21,6 +21,27 @@
 
 from .._constants import RAY_VERSION
 
+try:
+    from ray.data.datasource import FileExtensionFilter
+
+    class _FileFiler(FileExtensionFilter):  # pragma: no cover
+        def __init__(
+            self, file_extensions: Union[str, List[str]], exclude: Iterable[str]
+        ):
+            super().__init__(file_extensions, allow_if_no_extension=True)
+            self._exclude = set(exclude)
+
+        def _is_valid(self, path: str) -> bool:
+            return pathlib.Path(
+                path
+            ).name not in self._exclude and self._file_has_extension(path)
+
+        def __call__(self, paths: List[str]) -> List[str]:
+            return [path for path in paths if self._is_valid(path)]
+
+except ImportError:  # pragma: no cover
+    pass  # ray >=2.10
+
 
 class RayIO(object):
     def __init__(self, engine: ExecutionEngine):
@@ -248,17 +269,3 @@ def _read_json() -> RayDataFrame:  # pragma: no cover
 
     def _remote_args(self) -> Dict[str, Any]:
         return {"num_cpus": 1}
-
-
-class _FileFiler(FileExtensionFilter):  # pragma: no cover
-    def __init__(self, file_extensions: Union[str, List[str]], exclude: Iterable[str]):
-        super().__init__(file_extensions, allow_if_no_extension=True)
-        self._exclude = set(exclude)
-
-    def _is_valid(self, path: str) -> bool:
-        return pathlib.Path(
-            path
-        ).name not in self._exclude and self._file_has_extension(path)
-
-    def __call__(self, paths: List[str]) -> List[str]:
-        return [path for path in paths if self._is_valid(path)]