diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index e77ab626e..45ed2a171 100644 --- a/dvc/data_cloud.py +++ b/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, diff --git a/dvc/output/base.py b/dvc/output/base.py index 4a24d459c..faed0f5b2 100644 --- a/dvc/output/base.py +++ b/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 diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 204a81bba..67036ce61 100644 --- a/dvc/remote/base.py +++ b/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 = [ diff --git a/dvc/repo/add.py b/dvc/repo/add.py index 6041ccf9c..f5e6f658d 100644 --- a/dvc/repo/add.py +++ b/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) diff --git a/dvc/repo/imp_url.py b/dvc/repo/imp_url.py index 5b3f0b354..cbe83b55f 100644 --- a/dvc/repo/imp_url.py +++ b/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) diff --git a/dvc/stage/imports.py b/dvc/stage/imports.py index 3a5298033..8878220ae 100644 --- a/dvc/stage/imports.py +++ b/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):