output: introduce BaseOutput.transfer (#5861)

* output: introduce BaseOutput.transfer

* add: use BaseOutput.transfer for to-cache

dvc/data_cloud.py

@@ -92,29 +92,6 @@ class DataCloud:
             show_checksums=show_checksums,
         )
 
-    def transfer(
-        self, source, jobs=None, update=False, remote=None, command=None
-    ):
-        """Transfer data items in a cloud-agnostic way.
-
-        Args:
-            source (str): url for the source location.
-            jobs (int): number of jobs that can be running simultaneously.
-            update (bool): whether to update existing data or sync fully
-            remote (dvc.remote.base.BaseRemote): optional remote to compare
-                cache to. By default remote from core.remote config option
-                is used.
-            command (str): the command which is benefitting from this function
-                (to be used for reporting better error messages).
-        """
-        from dvc.fs import get_cloud_fs
-
-        from_fs = get_cloud_fs(self.repo, url=source)
-        remote = self.get_remote(remote, command)
-        return remote.transfer(
-            from_fs, from_fs.path_info, jobs=jobs, update=update
-        )
-
     def status(
         self,
         cache,

dvc/output/base.py

@@ -17,6 +17,7 @@ from dvc.exceptions import (
     RemoteCacheRequiredError,
 )
 from dvc.hash_info import HashInfo
+from dvc.objects import save as osave
 from dvc.objects.db import NamedCache
 from dvc.objects.errors import ObjectFormatError
 from dvc.objects.stage import stage as ostage
@@ -470,6 +471,42 @@ class BaseOutput:
         if self.scheme == "local" and self.use_scm_ignore:
             self.repo.scm.ignore(self.fspath)
 
+    def transfer(
+        self, source, odb=None, jobs=None, update=False, no_progress_bar=False,
+    ):
+        from dvc.fs import get_cloud_fs
+
+        if odb is None:
+            odb = self.odb
+
+        from_fs = get_cloud_fs(self.repo, url=source)
+        from_info = from_fs.path_info
+
+        # When running import-url --to-remote / add --to-remote/-o ... we
+        # assume that it is unlikely that the odb will contain majority of the
+        # hashes, so we transfer everything as is (even if that file might
+        # already be in the cache) and don't waste an upload to scan the layout
+        # of the source location. But when doing update --to-remote, there is
+        # a high probability that the odb might contain some of the hashes, so
+        # we first calculate all the hashes (but don't transfer anything) and
+        # then only update the missing cache files.
+
+        upload = not (update and from_fs.isdir(from_info))
+        jobs = jobs or min((from_fs.jobs, odb.fs.jobs))
+        obj = ostage(
+            odb,
+            from_info,
+            from_fs,
+            "md5",
+            upload=upload,
+            jobs=jobs,
+            no_progress_bar=no_progress_bar,
+        )
+        osave(odb, obj, jobs=jobs, move=upload)
+
+        self.hash_info = obj.hash_info
+        return obj
+
     def get_files_number(self, filter_info=None):
         if not self.use_cache or not self.hash_info:
             return 0

dvc/remote/base.py

@@ -8,8 +8,6 @@ from functools import partial, wraps
 
 from dvc.exceptions import DownloadError, UploadError
 from dvc.hash_info import HashInfo
-from dvc.objects import save
-from dvc.objects.stage import stage
 
 from ..progress import Tqdm
 from .index import RemoteIndex, RemoteIndexNoop
@@ -507,38 +505,6 @@ class Remote:
 
         return ret
 
-    def transfer(
-        self,
-        from_fs,
-        from_info,
-        jobs=None,
-        update=False,
-        no_progress_bar=False,
-    ):
-        # When running import-url --to-remote / add --to-remote/-o ... we
-        # assume that it is unlikely that the odb will contain majority of the
-        # hashes, so we transfer everything as is (even if that file might
-        # already be in the cache) and don't waste an upload to scan the layout
-        # of the source location. But when doing update --to-remote, there is
-        # a high probability that the odb might contain some of the hashes, so
-        # we first calculate all the hashes (but don't transfer anything) and
-        # then only update the missing cache files.
-
-        upload = not (update and from_fs.isdir(from_info))
-        jobs = jobs or min((from_fs.jobs, self.odb.fs.jobs))
-        obj = stage(
-            self.odb,
-            from_info,
-            from_fs,
-            "md5",
-            upload=upload,
-            jobs=jobs,
-            no_progress_bar=no_progress_bar,
-        )
-
-        save(self.odb, obj, jobs=jobs, move=upload)
-        return obj.hash_info
-
     @staticmethod
     def _log_missing_caches(hash_info_dict):
         missing_caches = [

dvc/repo/add.py

@@ -164,30 +164,14 @@ def _process_stages(
         (target,) = sub_targets
         (out,) = stage.outs
 
+        odb = None
         if to_remote:
-            out.hash_info = repo.cloud.transfer(
-                target,
-                jobs=kwargs.get("jobs"),
-                remote=kwargs.get("remote"),
-                command="add",
-            )
-        else:
-            from dvc.fs import get_cloud_fs
-            from dvc.objects import save as osave
-            from dvc.objects.stage import stage as ostage
+            remote = repo.cloud.get_remote(kwargs.get("remote"), "add")
+            odb = remote.odb
 
-            from_fs = get_cloud_fs(repo, url=target)
-            jobs = kwargs.get("jobs", min((from_fs.jobs, out.odb.fs.jobs)))
-            obj = ostage(
-                out.odb,
-                from_fs.path_info,
-                from_fs,
-                "md5",
-                upload=True,
-                jobs=jobs,
-            )
-            osave(out.odb, obj, jobs=jobs, move=False)
-            out.hash_info = obj.hash_info
+        out.transfer(target, odb=odb, jobs=kwargs.get("jobs"))
+
+        if to_cache:
             out.checkout()
 
         Dvcfile(repo, stage.path).dump(stage)

dvc/repo/imp_url.py

@@ -74,9 +74,8 @@ def imp_url(
     if no_exec:
         stage.ignore_outs()
     elif to_remote:
-        stage.outs[0].hash_info = self.cloud.transfer(
-            url, jobs=jobs, remote=remote, command="import-url"
-        )
+        remote = self.cloud.get_remote(remote, "import-url")
+        stage.outs[0].transfer(url, odb=remote.odb, jobs=jobs)
     else:
         stage.run(jobs=jobs)
 

dvc/stage/imports.py

@@ -13,9 +13,8 @@ def _update_import_on_remote(stage, remote, jobs):
         )
 
     url = stage.deps[0].def_path
-    stage.outs[0].hash_info = stage.repo.cloud.transfer(
-        url, jobs=jobs, update=True, remote=remote, command="update"
-    )
+    remote = stage.repo.cloud.get_remote(remote, "update")
+    stage.outs[0].transfer(url, odb=remote.odb, jobs=jobs, update=True)
 
 
 def update_import(stage, rev=None, to_remote=False, remote=None, jobs=None):