diff --git a/dvc/fs/base.py b/dvc/fs/base.py index e7c757bc5..99a06e929 100644 --- a/dvc/fs/base.py +++ b/dvc/fs/base.py @@ -187,7 +187,7 @@ class BaseFileSystem: def remove(self, path_info): raise RemoteActionNotImplemented("remove", self.scheme) - def makedirs(self, path_info): + def makedirs(self, path_info, **kwargs): """Optional: Implement only if the remote needs to create directories before copying/linking/moving data """ diff --git a/dvc/fs/fsspec_wrapper.py b/dvc/fs/fsspec_wrapper.py index 53d52889a..9b4a67de7 100644 --- a/dvc/fs/fsspec_wrapper.py +++ b/dvc/fs/fsspec_wrapper.py @@ -22,21 +22,11 @@ class FSSpecWrapper(BaseFileSystem): def fs(self): raise NotImplementedError - @lru_cache(512) def _with_bucket(self, path): - if isinstance(path, self.PATH_CLS): - return f"{path.bucket}/{path.path}" - return path + return str(path) def _strip_bucket(self, entry): - try: - bucket, path = entry.split("/", 1) - except ValueError: - # If there is no path attached, only returns - # the bucket (top-level). - bucket, path = entry, None - - return path or bucket + return entry def _strip_buckets(self, entries, detail=False): for entry in entries: @@ -86,6 +76,7 @@ class FSSpecWrapper(BaseFileSystem): return self.fs.open(self._with_bucket(path_info), mode=mode) def copy(self, from_info, to_info): + self.makedirs(to_info.parent) self.fs.copy(self._with_bucket(from_info), self._with_bucket(to_info)) def exists(self, path_info) -> bool: @@ -118,18 +109,25 @@ class FSSpecWrapper(BaseFileSystem): info["name"] = self._strip_bucket(info["name"]) return info + def makedirs(self, path_info, **kwargs): + self.fs.makedirs( + self._with_bucket(path_info), exist_ok=kwargs.pop("exist_ok", True) + ) + def _upload_fobj(self, fobj, to_info, size=None): + self.makedirs(to_info.parent) with self.open(to_info, "wb") as fdest: shutil.copyfileobj(fobj, fdest, length=fdest.blocksize) def _upload( self, from_file, to_info, name=None, no_progress_bar=False, **kwargs ): + self.makedirs(to_info.parent) size = os.path.getsize(from_file) with open(from_file, "rb") as fobj: self.upload_fobj( fobj, - self._with_bucket(to_info), + to_info, size=size, desc=name, no_progress_bar=no_progress_bar, @@ -158,6 +156,29 @@ class FSSpecWrapper(BaseFileSystem): class ObjectFSWrapper(FSSpecWrapper): TRAVERSE_PREFIX_LEN = 3 + @lru_cache(512) + def _with_bucket(self, path): + if isinstance(path, self.PATH_CLS): + return f"{path.bucket}/{path.path}" + return path + + def _strip_bucket(self, entry): + try: + bucket, path = entry.split("/", 1) + except ValueError: + # If there is no path attached, only returns + # the bucket (top-level). + bucket, path = entry, None + return path or bucket + + def makedirs(self, path_info, **kwargs): + # For object storages make this method a no-op. The original + # fs.makedirs() method will only check if the bucket exists + # and create if it doesn't though we don't want to support + # that behavior, and the check will cost some time so we'll + # simply ignore all mkdir()/makedirs() calls. + return None + def _isdir(self, path_info): # Directory in object storages are interpreted differently # among different fsspec providers, so this logic is a temporary diff --git a/dvc/fs/hdfs.py b/dvc/fs/hdfs.py index b2893d255..3e906340a 100644 --- a/dvc/fs/hdfs.py +++ b/dvc/fs/hdfs.py @@ -183,7 +183,7 @@ class HDFSFileSystem(BaseFileSystem): else: hdfs.delete_file(path_info.path) - def makedirs(self, path_info): + def makedirs(self, path_info, **kwargs): with self.hdfs(path_info) as hdfs: # NOTE: fs.create_dir creates parents by default hdfs.create_dir(path_info.path) diff --git a/dvc/fs/local.py b/dvc/fs/local.py index 147cb06fa..4c02b54d9 100644 --- a/dvc/fs/local.py +++ b/dvc/fs/local.py @@ -79,8 +79,8 @@ class LocalFileSystem(BaseFileSystem): raise NotImplementedError remove(path_info) - def makedirs(self, path_info): - makedirs(path_info, exist_ok=True) + def makedirs(self, path_info, **kwargs): + makedirs(path_info, exist_ok=kwargs.pop("exist_ok", True)) def isexec(self, path_info): mode = self.stat(path_info).st_mode diff --git a/dvc/fs/ssh/__init__.py b/dvc/fs/ssh/__init__.py index 496e25043..2fca225c2 100644 --- a/dvc/fs/ssh/__init__.py +++ b/dvc/fs/ssh/__init__.py @@ -173,7 +173,7 @@ class SSHFileSystem(BaseFileSystem): # pylint:disable=abstract-method with self.ssh() as ssh: ssh.remove(path_info.path) - def makedirs(self, path_info): + def makedirs(self, path_info, **kwargs): with self.ssh() as ssh: ssh.makedirs(path_info.path) diff --git a/dvc/fs/webdav.py b/dvc/fs/webdav.py index d7e5e11a7..7f9d6788a 100644 --- a/dvc/fs/webdav.py +++ b/dvc/fs/webdav.py @@ -81,10 +81,6 @@ class WebDAVFileSystem(FSSpecWrapper): # pylint:disable=abstract-method fobj, rpath, overwrite=True, size=size ) - def makedirs(self, path_info): - path = self.translate_path_info(path_info) - return self.fs.makedirs(path, exist_ok=True) - @lru_cache(512) def translate_path_info(self, path): if isinstance(path, self.PATH_CLS): @@ -93,9 +89,6 @@ class WebDAVFileSystem(FSSpecWrapper): # pylint:disable=abstract-method _with_bucket = translate_path_info - def _strip_bucket(self, entry): - return entry - class WebDAVSFileSystem(WebDAVFileSystem): # pylint:disable=abstract-method scheme = Schemes.WEBDAVS diff --git a/tests/func/test_fs.py b/tests/func/test_fs.py index bbfe95bf3..96154f9d7 100644 --- a/tests/func/test_fs.py +++ b/tests/func/test_fs.py @@ -1,3 +1,4 @@ +import io import os from operator import itemgetter from os.path import join @@ -324,7 +325,7 @@ def test_fs_ls(dvc, cloud): pytest.lazy_fixture("gdrive"), ], ) -def test_fs_find_recursive(dvc, cloud): +def test_fs_find(dvc, cloud): cloud.gen({"data": {"foo": "foo", "bar": {"baz": "baz"}, "quux": "quux"}}) cls, config, path_info = get_cloud_fs(dvc, **cloud.config) fs = cls(**config) @@ -382,3 +383,28 @@ def test_fs_fsspec_path_management(dvc, cloud): data_details = fs.info(data) assert data_details["name"].rstrip("/") == data.path assert data_details["type"] == "directory" + + +@pytest.mark.needs_internet +@pytest.mark.parametrize( + "cloud", + [ + pytest.lazy_fixture("azure"), + pytest.lazy_fixture("gs"), + pytest.lazy_fixture("s3"), + pytest.lazy_fixture("webdav"), + ], +) +def test_fs_makedirs_on_upload_and_copy(dvc, cloud): + cls, config, _ = get_cloud_fs(dvc, **cloud.config) + fs = cls(**config) + + with io.BytesIO(b"foo") as stream: + fs.upload_fobj(stream, cloud / "dir" / "foo") + + assert fs.isdir(cloud / "dir") + assert fs.exists(cloud / "dir" / "foo") + + fs.copy(cloud / "dir" / "foo", cloud / "dir2" / "foo") + assert fs.isdir(cloud / "dir2") + assert fs.exists(cloud / "dir2" / "foo")