From 61699d1d669159a320f5bb3022cefebf8fe7b9f5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Saugat=20Pachhai=20=28=E0=A4=B8=E0=A5=8C=E0=A4=97=E0=A4=BE?= =?UTF-8?q?=E0=A4=A4=29?= Date: Tue, 3 May 2022 10:09:41 +0545 Subject: [PATCH] fs: remove upload function --- dvc/data/stage.py | 2 +- dvc/fs/base.py | 23 ---------- dvc/fs/utils.py | 62 ++++++++++++++----------- dvc/repo/experiments/executor/ssh.py | 2 +- dvc/stage/cache.py | 67 +++++++++++++--------------- tests/func/objects/db/test_index.py | 2 +- tests/func/test_fs.py | 11 +++-- tests/func/test_remote.py | 18 -------- 8 files changed, 76 insertions(+), 111 deletions(-) diff --git a/dvc/data/stage.py b/dvc/data/stage.py index 04e2d1ad1..65d041b5e 100644 --- a/dvc/data/stage.py +++ b/dvc/data/stage.py @@ -41,7 +41,7 @@ def _upload_file(from_fs_path, fs, odb, upload_odb, callback=None): bytes=True, total=size, ) as cb: - upload_odb.fs.upload(stream, tmp_info, size=size, callback=cb) + upload_odb.fs.put_file(stream, tmp_info, size=size, callback=cb) odb.add(tmp_info, upload_odb.fs, stream.hash_info) meta = Meta(size=size) diff --git a/dvc/fs/base.py b/dvc/fs/base.py index fb923bd80..da96b2e39 100644 --- a/dvc/fs/base.py +++ b/dvc/fs/base.py @@ -362,29 +362,6 @@ class FileSystem: # pylint: enable=unused-argument - def upload( - self, - from_info: Union[AnyFSPath, IO], - to_info: AnyFSPath, - size: int = None, - callback: FsspecCallback = None, - ): - from .local import localfs - - if not hasattr(from_info, "read"): - logger.debug("Uploading '%s' to '%s'", from_info, to_info) - desc = localfs.path.name(from_info) - else: - desc = self.path.name(to_info) - - with FsspecCallback.as_tqdm_callback( - callback, - desc=desc, - bytes=True, - total=size or -1, - ) as cb: - return self.put_file(from_info, to_info, callback=cb, size=size) - def download( self, from_info: AnyFSPath, diff --git a/dvc/fs/utils.py b/dvc/fs/utils.py index 3c96791e4..d1cffe994 100644 --- a/dvc/fs/utils.py +++ b/dvc/fs/utils.py @@ -3,22 +3,22 @@ import logging import os from typing import TYPE_CHECKING, List, Optional +from ._callback import DEFAULT_CALLBACK, FsspecCallback from .base import RemoteActionNotImplemented from .local import LocalFileSystem if TYPE_CHECKING: - from .base import FileSystem - from .types import AnyPath + from .base import AnyFSPath, FileSystem logger = logging.getLogger(__name__) def _link( - link: "AnyPath", + link: "str", from_fs: "FileSystem", - from_path: "AnyPath", + from_path: "AnyFSPath", to_fs: "FileSystem", - to_path: "AnyPath", + to_path: "AnyFSPath", ) -> None: if not isinstance(from_fs, type(to_fs)): raise OSError(errno.EXDEV, "can't link across filesystems") @@ -34,36 +34,38 @@ def _link( ) from exc -def _copy( +def copy( from_fs: "FileSystem", - from_path: "AnyPath", + from_path: "AnyFSPath", to_fs: "FileSystem", - to_path: "AnyPath", + to_path: "AnyFSPath", + callback: "FsspecCallback" = DEFAULT_CALLBACK, ) -> None: if isinstance(from_fs, LocalFileSystem): - return to_fs.upload(from_path, to_path) + return to_fs.put_file(from_path, to_path, callback=callback) if isinstance(to_fs, LocalFileSystem): - return from_fs.download_file(from_path, to_path) + return from_fs.download_file(from_path, to_path, callback=callback) with from_fs.open(from_path, mode="rb") as fobj: size = from_fs.size(from_path) - return to_fs.upload(fobj, to_path, size=size) + return to_fs.put_file(fobj, to_path, size=size, callback=callback) def _try_links( - links: List["AnyPath"], + links: List["str"], from_fs: "FileSystem", - from_path: "AnyPath", + from_path: "AnyFSPath", to_fs: "FileSystem", - to_path: "AnyPath", + to_path: "AnyFSPath", + callback: "FsspecCallback" = DEFAULT_CALLBACK, ) -> None: error = None while links: link = links[0] if link == "copy": - return _copy(from_fs, from_path, to_fs, to_path) + return copy(from_fs, from_path, to_fs, to_path, callback=callback) try: return _link(link, from_fs, from_path, to_fs, to_path) @@ -81,11 +83,12 @@ def _try_links( def transfer( from_fs: "FileSystem", - from_path: "AnyPath", + from_path: "AnyFSPath", to_fs: "FileSystem", - to_path: "AnyPath", + to_path: "AnyFSPath", hardlink: bool = False, - links: Optional[List["AnyPath"]] = None, + links: Optional[List["str"]] = None, + callback: "FsspecCallback" = None, ) -> None: try: assert not (hardlink and links) @@ -93,7 +96,14 @@ def transfer( links = links or ["reflink", "hardlink", "copy"] else: links = links or ["reflink", "copy"] - _try_links(links, from_fs, from_path, to_fs, to_path) + + with FsspecCallback.as_tqdm_callback( + callback, + desc=from_fs.path.name(from_path), + bytes=True, + total=-1, + ) as cb: + _try_links(links, from_fs, from_path, to_fs, to_path, callback=cb) except OSError as exc: # If the target file already exists, we are going to simply # ignore the exception (#4992). @@ -114,11 +124,11 @@ def transfer( def _test_link( - link: "AnyPath", + link: "str", from_fs: "FileSystem", - from_file: "AnyPath", + from_file: "AnyFSPath", to_fs: "FileSystem", - to_file: "AnyPath", + to_file: "AnyFSPath", ) -> bool: try: _try_links([link], from_fs, from_file, to_fs, to_file) @@ -136,12 +146,12 @@ def _test_link( def test_links( - links: List["AnyPath"], + links: List["str"], from_fs: "FileSystem", - from_path: "AnyPath", + from_path: "AnyFSPath", to_fs: "FileSystem", - to_path: "AnyPath", -) -> List["AnyPath"]: + to_path: "AnyFSPath", +) -> List["AnyFSPath"]: from dvc.utils import tmp_fname from_file = from_fs.path.join(from_path, tmp_fname()) diff --git a/dvc/repo/experiments/executor/ssh.py b/dvc/repo/experiments/executor/ssh.py index 29fc79d83..adc9cc7f1 100644 --- a/dvc/repo/experiments/executor/ssh.py +++ b/dvc/repo/experiments/executor/ssh.py @@ -175,7 +175,7 @@ class SSHExecutor(BaseExecutor): posixpath.join(self._repo_abspath, self.dvc_dir) ) assert self._setup_script - fs.upload(self._setup_script, script_path) + fs.put_file(self._setup_script, script_path) def _ssh_cmd(self, sshfs, cmd, chdir=None, **kwargs): working_dir = chdir or self.root_dir diff --git a/dvc/stage/cache.py b/dvc/stage/cache.py index 04e225f93..e4e922546 100644 --- a/dvc/stage/cache.py +++ b/dvc/stage/cache.py @@ -6,11 +6,13 @@ from typing import TYPE_CHECKING, Optional from funcy import cached_property, first +from dvc import fs +from dvc.data.transfer import _log_exceptions from dvc.exceptions import DvcException from dvc.utils import dict_sha256, relpath if TYPE_CHECKING: - from dvc.objects.db.base import ObjectDB + from dvc.objects.db import ObjectDB logger = logging.getLogger(__name__) @@ -210,51 +212,46 @@ class StageCache: ) cached_stage.checkout() - @staticmethod - def _transfer(func, from_remote, to_remote): + def transfer(self, from_odb, to_odb): + from dvc.fs._callback import FsspecCallback + + from_fs = from_odb.fs + to_fs = from_odb.fs + func = _log_exceptions(fs.utils.copy) + runs = from_fs.path.join(from_odb.fs_path, "runs") + ret = [] + if not from_fs.exists(runs): + return ret - runs = from_remote.fs.path.join(from_remote.fs_path, "runs") - if not from_remote.fs.exists(runs): - return [] - - from_path = from_remote.fs.path - for src in from_remote.fs.find(runs): - rel = from_path.relpath(src, from_remote.fs_path) - dst = to_remote.fs.path.join(to_remote.fs_path, rel) - key = to_remote.fs.path.parent(dst) + for src in from_fs.find(runs): + rel = from_fs.path.relpath(src, from_odb.fs_path) + dst = to_fs.path.join(to_odb.fs_path, rel) + key = to_fs.path.parent(dst) # check if any build cache already exists for this key # TODO: check if MaxKeys=1 or something like that applies # or otherwise this will take a lot of time! - if to_remote.fs.exists(key) and first(to_remote.fs.find(key)): + if to_fs.exists(key) and first(to_fs.find(key)): continue - func(src, dst) - ret.append( - (from_path.name(from_path.parent(src)), from_path.name(src)) - ) + src_name = from_fs.path.name(src) + parent_name = from_fs.path.name(from_fs.path.parent(src)) + with FsspecCallback.as_tqdm_callback( + desc=src_name, + bytes=True, + total=-1, + ) as cb: + func(from_fs, src, to_fs, dst, callback=cb) + ret.append((parent_name, src_name)) return ret def push(self, remote: Optional[str], odb: Optional["ObjectDB"] = None): - from dvc.data.transfer import _log_exceptions + dest_odb = odb or self.repo.cloud.get_remote_odb(remote) + return self.transfer(self.repo.odb.local, dest_odb) - if odb is None: - odb = self.repo.cloud.get_remote_odb(remote) - return self._transfer( - _log_exceptions(odb.fs.upload), - self.repo.odb.local, - odb, - ) - - def pull(self, remote: Optional[str]): - from dvc.data.transfer import _log_exceptions - - odb = self.repo.cloud.get_remote_odb(remote) - return self._transfer( - _log_exceptions(odb.fs.download), - odb, - self.repo.odb.local, - ) + def pull(self, remote: Optional[str], odb: Optional["ObjectDB"] = None): + odb = odb or self.repo.cloud.get_remote_odb(remote) + return self.transfer(odb, self.repo.odb.local) def get_used_objs(self, used_run_cache, *args, **kwargs): """Return used cache for the specified run-cached stages.""" diff --git a/tests/func/objects/db/test_index.py b/tests/func/objects/db/test_index.py index 6df598fe5..4aa6d3ca3 100644 --- a/tests/func/objects/db/test_index.py +++ b/tests/func/objects/db/test_index.py @@ -76,7 +76,7 @@ def test_partial_upload(tmp_dir, dvc, index, mocker): tmp_dir.dvc_gen({"foo": "foo content"}) tmp_dir.dvc_gen({"bar": {"baz": "baz content"}}) - original = LocalFileSystem.upload + original = LocalFileSystem.put_file def unreliable_upload(self, from_file, to_info, name=None, **kwargs): if "baz" in name: diff --git a/tests/func/test_fs.py b/tests/func/test_fs.py index 8988c2237..d40227012 100644 --- a/tests/func/test_fs.py +++ b/tests/func/test_fs.py @@ -1,5 +1,5 @@ -import io import os +from io import BytesIO from operator import itemgetter from os.path import join @@ -229,9 +229,8 @@ def test_fs_upload_fobj(dvc, tmp_dir, cloud): 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(stream, (cloud / "dir" / "foo").fs_path) + with BytesIO(b"foo") as stream: + fs.put_file(stream, (cloud / "dir" / "foo").fs_path) assert fs.isdir((cloud / "dir").fs_path) assert fs.exists((cloud / "dir" / "foo").fs_path) @@ -249,7 +248,7 @@ def test_upload_callback(tmp_dir, dvc, cloud): expected_size = os.path.getsize(tmp_dir / "foo") callback = FsspecCallback() - fs.upload( + fs.put_file( (tmp_dir / "foo").fs_path, (cloud / "foo").fs_path, callback=callback, @@ -264,7 +263,7 @@ def test_download_callback(tmp_dir, dvc, cloud, local_cloud): fs = cls(**config) (tmp_dir / "to_upload").write_text("foo") - fs.upload((tmp_dir / "to_upload").fs_path, (cloud / "foo").fs_path) + fs.put_file((tmp_dir / "to_upload").fs_path, (cloud / "foo").fs_path) expected_size = fs.size((cloud / "foo").fs_path) callback = FsspecCallback() diff --git a/tests/func/test_remote.py b/tests/func/test_remote.py index ff844ed7e..dbfe080f9 100644 --- a/tests/func/test_remote.py +++ b/tests/func/test_remote.py @@ -430,21 +430,3 @@ def test_push_incomplete_dir(tmp_dir, dvc, mocker, local_remote): assert not remote_odb.exists(out.hash_info) assert not remote_odb.exists(file_objs[0]) assert remote_odb.exists(file_objs[1]) - - -def test_upload_exists(tmp_dir, dvc, local_remote): - tmp_dir.gen("foo", "foo") - odb = dvc.cloud.get_remote_odb("upstream") - # allow uploaded files to be writable for this test, - # normally they are set to read-only for DVC remotes - odb.fs.CACHE_MODE = 0o644 - - from_info = (tmp_dir / "foo").fs_path - to_info = odb.fs.path.join(odb.fs_path, "foo") - odb.fs.upload(from_info, to_info) - assert odb.fs.exists(to_info) - - tmp_dir.gen("foo", "bar") - odb.fs.upload(from_info, to_info) - with odb.fs.open(to_info) as fobj: - assert fobj.read() == "bar"