make fsspec preferred backend in etils

etils/epath/backend.py

@@ -456,8 +456,17 @@ def isdir(self, path: PathLike) -> bool:
     return self.fs(path).isdir(path)
 
   def listdir(self, path: PathLike) -> list[str]:
-    paths = self.fs(path).listdir(path, detail=False)
-    return [os.path.basename(p) for p in paths if not p.endswith('~')]
+    fs = self.fs(path)
+    stripped_path = fs._strip_protocol(os.fspath(path)).rstrip('/')
+    paths = fs.listdir(path, detail=False)
+    # Filter out the directory itself (if returned by fsspec) to align with
+    # TensorFlow backend behavior.
+    return [
+        os.fspath(os.path.basename(p))
+        for p in paths
+        if fs._strip_protocol(p).rstrip('/') != stripped_path
+        and not p.endswith('~')
+    ]
 
   def glob(self, path: PathLike) -> list[str]:
     protocol = _get_protocol(path)

etils/epath/backend_test.py

@@ -831,3 +831,26 @@ def _escape(s):
         expected_copy_overwrite=IsADirectoryError(),
     ),
 )
+
+
+def test_fsspec_gcs_listdir_filtering():
+  from unittest import mock
+  import fsspec
+
+  with mock.patch.object(fsspec, 'filesystem') as mock_filesystem:
+    mock_fs = mock.MagicMock()
+    mock_fs.listdir.return_value = [
+        'bucket/dir',
+        'bucket/dir/file.txt',
+        'bucket/dir/subdir',
+    ]
+    mock_fs._strip_protocol.side_effect = lambda p: p.replace('gs://', '')
+    mock_filesystem.return_value = mock_fs
+
+    backend = epath.backend.fsspec_backend
+    backend._get_filesystem.cache_clear()
+
+    files = backend.listdir('gs://bucket/dir')
+
+    assert sorted(files) == ['file.txt', 'subdir']
+    mock_fs.listdir.assert_called_once_with('gs://bucket/dir', detail=False)

etils/epath/gpath.py

@@ -18,6 +18,7 @@
 
 import functools
 import importlib.util
+import logging
 import ntpath
 import os
 import pathlib
@@ -55,6 +56,15 @@ def _epath_use_tf() -> bool:
       '0',
   ]
 
+
+def _epath_prefer_fsspec() -> bool:
+  return os.environ.get('EPATH_PREFER_FSSPEC', '').lower() in [
+      'true',
+      'yes',
+      'y',
+      '1',
+  ]
+
 _PREFIX_TO_BACKEND = {
     'gs': backend_lib.fsspec_backend,
     's3': backend_lib.fsspec_backend,
@@ -67,6 +77,18 @@ def _epath_use_tf() -> bool:
     backend_lib.tf_backend,
 })
 
+
+_SCHEME_TO_LOGGED_BACKEND = {}
+
+def _log_backend_selection(scheme: str, backend: backend_lib.Backend):
+  backend_type = type(backend)
+  if _SCHEME_TO_LOGGED_BACKEND.get(scheme) != backend_type:
+    logging.info(
+        'epath: Using %s backend for %s://', backend_type.__name__, scheme
+    )
+    _SCHEME_TO_LOGGED_BACKEND[scheme] = backend_type
+
+
 # Available modes (from tensorflow/python/lib/io/file_io.py;l=55)
 # Also exclude `+` as broken in gfile
 _OPEN_MODES = ('r', 'w', 'a')
@@ -112,11 +134,12 @@ def _uri_scheme(self) -> Optional[str]:
   def _backend(self) -> backend_lib.Backend:
     try:
       backend = _PREFIX_TO_BACKEND[self._uri_scheme]
-      # Choose tf_backend if tf is installed. We don't use FSSpec by default
-      # for retro-compatibility, because needed dependencies (gcsfs or s3fs)
-      # may not be installed. fsspec_backend was indeed introduced later.
-      if _is_tf_installed() and self._uri_scheme is not None:
-        return backend_lib.tf_backend
+      if self._uri_scheme is not None:
+        if _is_fsspec_gcsfs_installed():
+          backend = backend_lib.fsspec_backend
+        elif _is_tf_installed():
+          backend = backend_lib.tf_backend
+        _log_backend_selection(self._uri_scheme, backend)
       return backend
     except KeyError:
       supported = ', '.join(f'`{k}://`' for k in _PREFIX_TO_BACKEND)
@@ -308,6 +331,17 @@ class WindowsGPath(pathlib.PureWindowsPath, _GPath):
   _PATH = ntpath
 
 
+@functools.cache
+def _is_fsspec_gcsfs_installed() -> bool:
+  """Checks whether fsspec and gcsfs are installed and environment variable is set"""
+  if not _epath_prefer_fsspec():
+    return False
+  return (
+      importlib.util.find_spec('fsspec') is not None
+      and importlib.util.find_spec('gcsfs') is not None
+  )
+
+
 @functools.cache
 def _is_tf_installed() -> bool:
   """Checks whether TensorFlow is installed."""

etils/epath/gpath_test.py

@@ -434,20 +434,73 @@ def test_use_backend():
   gs_path = epath.Path('gs://tfds-data/datasets')
   loc_path = epath.Path('/local/tfds-data/datasets')
 
-  gs_backend = epath.backend.tf_backend  # pylint: disable=unused-variable
-  loc_backend = epath.backend.os_backend  # pylint: disable=unused-variable
-  with mock.patch.object(epath.gpath, '_is_tf_installed', return_value=True):
-    assert epath.gpath._get_backend(gs_path, gs_path) == gs_backend
-  with mock.patch.object(epath.gpath, '_is_tf_installed', return_value=False):
-    assert (
-        epath.gpath._get_backend(gs_path, gs_path)
-        == epath.backend.fsspec_backend
-    )
-
-  assert epath.gpath._get_backend(gs_path, gs_path) == gs_backend  # pytype: disable=wrong-arg-types
-  assert epath.gpath._get_backend(gs_path, loc_path) == gs_backend  # pytype: disable=wrong-arg-types
-  assert epath.gpath._get_backend(loc_path, gs_path) == gs_backend  # pytype: disable=wrong-arg-types
-  assert epath.gpath._get_backend(loc_path, loc_path) == loc_backend  # pytype: disable=wrong-arg-types
+  tf_backend = epath.backend.tf_backend
+  fsspec_backend = epath.backend.fsspec_backend
+  loc_backend = epath.backend.os_backend
+
+  # Case 1: fsspec+gcsfs is available and TF is installed -> prefer fsspec
+  with mock.patch.object(
+      epath.gpath, '_is_fsspec_gcsfs_installed', return_value=True
+  ), mock.patch.object(epath.gpath, '_is_tf_installed', return_value=True):
+    assert epath.gpath._get_backend(gs_path, gs_path) == fsspec_backend
+    assert epath.gpath._get_backend(gs_path, loc_path) == fsspec_backend
+    assert epath.gpath._get_backend(loc_path, gs_path) == fsspec_backend
+    assert epath.gpath._get_backend(loc_path, loc_path) == loc_backend
+
+  # Case 2: Only fsspec+gcsfs is available -> use fsspec
+  with mock.patch.object(
+      epath.gpath, '_is_fsspec_gcsfs_installed', return_value=True
+  ), mock.patch.object(epath.gpath, '_is_tf_installed', return_value=False):
+    assert epath.gpath._get_backend(gs_path, gs_path) == fsspec_backend
+    assert epath.gpath._get_backend(gs_path, loc_path) == fsspec_backend
+    assert epath.gpath._get_backend(loc_path, gs_path) == fsspec_backend
+    assert epath.gpath._get_backend(loc_path, loc_path) == loc_backend
+
+  # Case 3: Only TF is installed -> use TF
+  with mock.patch.object(
+      epath.gpath, '_is_fsspec_gcsfs_installed', return_value=False
+  ), mock.patch.object(epath.gpath, '_is_tf_installed', return_value=True):
+    assert epath.gpath._get_backend(gs_path, gs_path) == tf_backend
+    assert epath.gpath._get_backend(gs_path, loc_path) == tf_backend
+    assert epath.gpath._get_backend(loc_path, gs_path) == tf_backend
+    assert epath.gpath._get_backend(loc_path, loc_path) == loc_backend
+
+  # Case 4: Neither is available -> use fsspec (default fallback)
+  with mock.patch.object(
+      epath.gpath, '_is_fsspec_gcsfs_installed', return_value=False
+  ), mock.patch.object(epath.gpath, '_is_tf_installed', return_value=False):
+    assert epath.gpath._get_backend(gs_path, gs_path) == fsspec_backend
+    assert epath.gpath._get_backend(gs_path, loc_path) == fsspec_backend
+    assert epath.gpath._get_backend(loc_path, gs_path) == fsspec_backend
+    assert epath.gpath._get_backend(loc_path, loc_path) == loc_backend
+
+
+def test_is_fsspec_gcsfs_installed():
+  # We need to clear the cache because the function is cached
+  epath.gpath._is_fsspec_gcsfs_installed.cache_clear()
+
+  # Case 1: Toggle is False -> should be False regardless of installation
+  with mock.patch.object(
+      epath.gpath, '_epath_prefer_fsspec', return_value=False
+  ):
+    assert not epath.gpath._is_fsspec_gcsfs_installed()
+
+  # Case 2: Toggle is True, but packages missing -> False
+  epath.gpath._is_fsspec_gcsfs_installed.cache_clear()
+  with mock.patch.object(
+      epath.gpath, '_epath_prefer_fsspec', return_value=True
+  ), mock.patch('importlib.util.find_spec', return_value=None):
+    assert not epath.gpath._is_fsspec_gcsfs_installed()
+
+  # Case 3: Toggle is True, and packages present -> True
+  epath.gpath._is_fsspec_gcsfs_installed.cache_clear()
+  with mock.patch.object(
+      epath.gpath, '_epath_prefer_fsspec', return_value=True
+  ), mock.patch('importlib.util.find_spec', return_value=mock.MagicMock()):
+    assert epath.gpath._is_fsspec_gcsfs_installed()
+
+  # Clean up cache after test
+  epath.gpath._is_fsspec_gcsfs_installed.cache_clear()
 
 
 @epy.testing.non_hermetic
@@ -463,3 +516,19 @@ def test_relative_to():
   assert path.relative_to('gs://bucket/dir') == epath.Path('subdir')
   with pytest.raises(ValueError, match='not in the subpath'):
     path.relative_to('gs://bucket/other-dir')
+
+
+def test_epath_prefer_fsspec_env():
+  # Without env var, it should be False by default (opt-in)
+  with mock.patch.dict(os.environ, {}, clear=True):
+    assert not epath.gpath._epath_prefer_fsspec()
+
+  # With env var set to true/yes/y/1, it should be True
+  for val in ['true', 'True', 'YES', 'y', '1']:
+    with mock.patch.dict(os.environ, {'EPATH_PREFER_FSSPEC': val}):
+      assert epath.gpath._epath_prefer_fsspec()
+
+  # With env var set to other values, it should be False
+  for val in ['false', 'False', 'NO', 'n', '0', 'random']:
+    with mock.patch.dict(os.environ, {'EPATH_PREFER_FSSPEC': val}):
+      assert not epath.gpath._epath_prefer_fsspec()