fsspec: split the cloud-specific functionality from the base wrapper (#6014)

* fsspec: split the cloud-specific functionality from the base fs

* rename Cloud to Object

dvc/fs/azure.py

@@ -12,7 +12,7 @@ from dvc.path_info import CloudURLInfo
 from dvc.scheme import Schemes
 from dvc.utils import format_link
 
-from .fsspec_wrapper import FSSpecWrapper
+from .fsspec_wrapper import ObjectFSWrapper
 
 logger = logging.getLogger(__name__)
 _DEFAULT_CREDS_STEPS = (
@@ -67,7 +67,8 @@ class AzureAuthError(DvcException):
     pass
 
 
-class AzureFileSystem(FSSpecWrapper):  # pylint:disable=abstract-method
+# pylint:disable=abstract-method
+class AzureFileSystem(ObjectFSWrapper):
     scheme = Schemes.AZURE
     PATH_CLS = CloudURLInfo
     PARAM_CHECKSUM = "etag"

dvc/fs/fsspec_wrapper.py

@@ -57,28 +57,21 @@ class FSSpecWrapper(BaseFileSystem):
         host filesystem"""
         return {}
 
-    def isdir(self, path_info):
-        if not self.exists(path_info):
-            return False
+    def _isdir(self, path_info):
+        return self.fs.isdir(self._with_bucket(path_info))
 
-        # Directory in object storages are interpreted differently
-        # among different fsspec providers, so this logic is a temporary
-        # measure for us to adapt as of now. It checks whether it is a
-        # directory (as in a prefix with contents) or whether it is an empty
-        # file where it's name ends with a forward slash
-        entry = self.info(path_info)
-        return entry["type"] == "directory" or (
-            entry["size"] == 0
-            and entry["type"] == "file"
-            and entry["name"].endswith("/")
-        )
+    def isdir(self, path_info):
+        try:
+            return self._isdir(path_info)
+        except FileNotFoundError:
+            return False
 
     def isfile(self, path_info):
-        if not self.exists(path_info):
+        try:
+            return not self._isdir(path_info)
+        except FileNotFoundError:
             return False
 
-        return not self.isdir(path_info)
-
     def is_empty(self, path_info):
         entry = self.info(path_info)
         if entry["type"] == "directory":
@@ -107,14 +100,6 @@ class FSSpecWrapper(BaseFileSystem):
         if detail:
             files = files.values()
 
-        # When calling find() on a file, it returns the same file in a list.
-        # For object-based storages, the same behavior applies to empty
-        # directories since they are represented as files. This condition
-        # checks whether we should yield an empty list (if it is an empty
-        # directory) or just yield the file itself.
-        if len(files) == 1 and files[0] == path and self.isdir(path_info):
-            return None
-
         yield from self._strip_buckets(files, detail=detail)
 
     def walk_files(self, path_info, **kwargs):
@@ -164,3 +149,36 @@ class FSSpecWrapper(BaseFileSystem):
             ) as wrapped:
                 with open(to_file, "wb") as fdest:
                     shutil.copyfileobj(wrapped, fdest, length=fobj.blocksize)
+
+
+# pylint: disable=abstract-method
+class ObjectFSWrapper(FSSpecWrapper):
+    def _isdir(self, path_info):
+        # Directory in object storages are interpreted differently
+        # among different fsspec providers, so this logic is a temporary
+        # measure for us to adapt as of now. It checks whether it is a
+        # directory (as in a prefix with contents) or whether it is an empty
+        # file where it's name ends with a forward slash
+
+        entry = self.info(path_info)
+        return entry["type"] == "directory" or (
+            entry["size"] == 0
+            and entry["type"] == "file"
+            and entry["name"].endswith("/")
+        )
+
+    def find(self, path_info, detail=False):
+        path = self._with_bucket(path_info)
+        files = self.fs.find(path, detail=detail)
+        if detail:
+            files = files.values()
+
+        # When calling find() on a file, it returns the same file in a list.
+        # For object-based storages, the same behavior applies to empty
+        # directories since they are represented as files. This condition
+        # checks whether we should yield an empty list (if it is an empty
+        # directory) or just yield the file itself.
+        if len(files) == 1 and files[0] == path and self.isdir(path_info):
+            return None
+
+        yield from self._strip_buckets(files, detail=detail)

dvc/fs/gs.py

@@ -6,10 +6,11 @@ from funcy import cached_property, wrap_prop
 from dvc.path_info import CloudURLInfo
 from dvc.scheme import Schemes
 
-from .fsspec_wrapper import FSSpecWrapper
+from .fsspec_wrapper import ObjectFSWrapper
 
 
-class GSFileSystem(FSSpecWrapper):  # pylint:disable=abstract-method
+# pylint:disable=abstract-method
+class GSFileSystem(ObjectFSWrapper):
     scheme = Schemes.GS
     PATH_CLS = CloudURLInfo
     REQUIRES = {"gcsfs": "gcsfs"}

dvc/fs/s3.py

@@ -9,12 +9,13 @@ from dvc.path_info import CloudURLInfo
 from dvc.progress import Tqdm
 from dvc.scheme import Schemes
 
-from .fsspec_wrapper import FSSpecWrapper
+from .fsspec_wrapper import ObjectFSWrapper
 
 _AWS_CONFIG_PATH = os.path.join(os.path.expanduser("~"), ".aws", "config")
 
 
-class BaseS3FileSystem(FSSpecWrapper):  # pylint:disable=abstract-method
+# pylint:disable=abstract-method
+class BaseS3FileSystem(ObjectFSWrapper):
     scheme = Schemes.S3
     PATH_CLS = CloudURLInfo
     REQUIRES = {"s3fs": "s3fs", "boto3": "boto3"}