From 960ece926ce635837f49fe848ff291350b84061d Mon Sep 17 00:00:00 2001 From: Batuhan Taskaya Date: Wed, 19 May 2021 18:12:07 +0300 Subject: [PATCH] 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 | 5 +-- dvc/fs/fsspec_wrapper.py | 68 +++++++++++++++++++++++++--------------- dvc/fs/gs.py | 5 +-- dvc/fs/s3.py | 5 +-- 4 files changed, 52 insertions(+), 31 deletions(-) diff --git a/dvc/fs/azure.py b/dvc/fs/azure.py index c3ddb1078..fb7132e48 100644 --- a/dvc/fs/azure.py +++ b/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" diff --git a/dvc/fs/fsspec_wrapper.py b/dvc/fs/fsspec_wrapper.py index 6e13c9c3f..b970be6b2 100644 --- a/dvc/fs/fsspec_wrapper.py +++ b/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) diff --git a/dvc/fs/gs.py b/dvc/fs/gs.py index 81961089d..a375ac50c 100644 --- a/dvc/fs/gs.py +++ b/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"} diff --git a/dvc/fs/s3.py b/dvc/fs/s3.py index 0a6b6165c..7dc74c0fa 100644 --- a/dvc/fs/s3.py +++ b/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"}