fetch: support passing config

For now not exposing it in CLI since there is no need/use yet.

Requirement for https://github.com/iterative/studio/pull/6541

dvc/cachemgr.py

@@ -24,7 +24,7 @@ def _get_odb(
     if not settings:
         return None
 
-    cls, config, fs_path = get_cloud_fs(repo, **settings)
+    cls, config, fs_path = get_cloud_fs(repo.config, **settings)
     fs = fs or cls(**config)
     if prefix:
         fs_path = fs.path.join(fs_path, *prefix)

dvc/data_cloud.py

@@ -90,7 +90,7 @@ class DataCloud:
         if name:
             from dvc.fs import get_cloud_fs
 
-            cls, config, fs_path = get_cloud_fs(self.repo, name=name)
+            cls, config, fs_path = get_cloud_fs(self.repo.config, name=name)
 
             if config.get("worktree"):
                 version_aware = config.get("version_aware")

dvc/dependency/base.py

@@ -50,7 +50,7 @@ class Dependency(Output):
             self.fs_path = self.fs.path.version_path(self.fs_path, self.meta.version_id)
 
     def download(self, to, jobs=None):
-        fs_download(self.fs, self.fs_path, to, jobs=jobs)
+        fs_download(self.fs, self.fs_path, to.fs_path, jobs=jobs)
 
     def save(self):
         super().save()

dvc/fs/__init__.py

@@ -1,3 +1,4 @@
+from typing import Optional
 from urllib.parse import urlparse
 
 from dvc_http import HTTPFileSystem, HTTPSFileSystem  # noqa: F401
@@ -18,7 +19,7 @@ from dvc_objects.fs import (  # noqa: F401
     system,
     utils,
 )
-from dvc_objects.fs.base import AnyFSPath, FileSystem  # noqa: F401
+from dvc_objects.fs.base import AnyFSPath, FileSystem  # noqa: F401, TCH001
 from dvc_objects.fs.errors import (  # noqa: F401
     AuthError,
     ConfigError,
@@ -48,18 +49,45 @@ known_implementations.update(
 # pylint: enable=unused-import
 
 
-def download(fs, fs_path, to, jobs=None):
+def download(fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None):
     with Callback.as_tqdm_callback(
         desc=f"Downloading {fs.path.name(fs_path)}",
         unit="files",
     ) as cb:
-        fs.get(fs_path, to.fs_path, batch_size=jobs, callback=cb)
+        # NOTE: We use dvc-objects generic.copy over fs.get since it makes file
+        # download atomic and avoids fsspec glob/regex path expansion.
+        if fs.isdir(fs_path):
+            from_infos = [
+                path
+                for path in fs.find(fs_path)
+                if not path.endswith(fs.path.flavour.sep)
+            ]
+            if not from_infos:
+                return localfs.makedirs(to, exist_ok=True)
+            to_infos = [
+                localfs.path.join(to, *fs.path.relparts(info, fs_path))
+                for info in from_infos
+            ]
+        else:
+            from_infos = [fs_path]
+            to_infos = [to]
+
+        cb.set_size(len(from_infos))
+        jobs = jobs or fs.jobs
+        generic.copy(
+            fs,
+            from_infos,
+            localfs,
+            to_infos,
+            callback=cb,
+            batch_size=jobs,
+        )
 
 
 def parse_external_url(url, config=None):
     remote_config = dict(config) if config else {}
     remote_config["url"] = url
-    fs_cls, fs_config, fs_path = get_cloud_fs(None, **remote_config)
+    fs_cls, fs_config, fs_path = get_cloud_fs({}, **remote_config)
     fs = fs_cls(**fs_config)
     return fs, fs_path
 
@@ -105,18 +133,14 @@ def _resolve_remote_refs(config, remote_conf):
         return remote_conf
 
     base = get_fs_config(config, name=parsed.netloc)
-    cls, _, _ = _get_cloud_fs(config, **base)
+    cls, _, _ = get_cloud_fs(config, **base)
     relpath = parsed.path.lstrip("/").replace("/", cls.sep)
     url = cls.sep.join((base["url"], relpath))
     return {**base, **remote_conf, "url": url}
 
 
-def get_cloud_fs(repo, **kwargs):
-    repo_config = repo.config if repo else {}
-    return _get_cloud_fs(repo_config, **kwargs)
-
-
-def _get_cloud_fs(repo_config, **kwargs):
+def get_cloud_fs(repo_config, **kwargs):
+    repo_config = repo_config or {}
     core_config = repo_config.get("core", {})
 
     remote_conf = get_fs_config(repo_config, **kwargs)

dvc/output.py

@@ -354,7 +354,11 @@ class Output:
         if fs_config is not None:
             fs_kwargs.update(**fs_config)
 
-        fs_cls, fs_config, fs_path = get_cloud_fs(self.repo, url=path, **fs_kwargs)
+        fs_cls, fs_config, fs_path = get_cloud_fs(
+            self.repo.config if self.repo else {},
+            url=path,
+            **fs_kwargs,
+        )
         self.fs = fs_cls(**fs_config)
 
         if (
@@ -1017,7 +1021,9 @@ class Output:
         if odb is None:
             odb = self.cache
 
-        cls, config, from_info = get_cloud_fs(self.repo, url=source)
+        cls, config, from_info = get_cloud_fs(
+            self.repo.config if self.repo else {}, url=source
+        )
         from_fs = cls(**config)
 
         # When running import-url --to-remote / add --to-remote/-o ... we

dvc/repo/__init__.py

@@ -146,7 +146,6 @@ class Repo:
         scm: Optional[Union["Git", "NoSCM"]] = None,
     ):
         from dvc.cachemgr import CacheManager
-        from dvc.config import Config
         from dvc.data_cloud import DataCloud
         from dvc.fs import GitFileSystem, LocalFileSystem, localfs
         from dvc.lock import LockNoop, make_lock
@@ -163,6 +162,7 @@ class Repo:
         self._fs_conf = {"repo_factory": repo_factory}
         self._fs = fs or localfs
         self._scm = scm
+        self._config = config
         self._data_index = None
 
         if rev and not fs:
@@ -182,7 +182,6 @@ class Repo:
             scm=scm,
         )
 
-        self.config: Config = Config(self.dvc_dir, fs=self.fs, config=config)
         self._uninitialized = uninitialized
 
         # used by DVCFileSystem to determine if it should traverse subrepos
@@ -234,6 +233,12 @@ class Repo:
     def __str__(self):
         return self.url or self.root_dir
 
+    @cached_property
+    def config(self):
+        from dvc.config import Config
+
+        return Config(self.dvc_dir, fs=self.fs, config=self._config)
+
     @cached_property
     def local_dvc_dir(self):
         from dvc.fs import GitFileSystem, LocalFileSystem
@@ -629,6 +634,7 @@ class Repo:
         self.__dict__.pop("dvcignore", None)
         self.__dict__.pop("dvcfs", None)
         self.__dict__.pop("datafs", None)
+        self.__dict__.pop("config", None)
 
     def __enter__(self):
         return self

dvc/repo/brancher.py

@@ -67,6 +67,7 @@ def brancher(  # noqa: E302
 
     saved_fs = self.fs
     saved_root = self.root_dir
+    saved_dvc_dir = self.dvc_dir
 
     scm = self.scm
 
@@ -101,6 +102,8 @@ def brancher(  # noqa: E302
     finally:
         self.fs = saved_fs
         self.root_dir = saved_root
+        self.dvc_dir = saved_dvc_dir
+        self._reset()  # pylint: disable=protected-access
 
 
 def _switch_fs(
@@ -125,6 +128,8 @@ def _switch_fs(
 
     repo.fs = fs
     repo.root_dir = root_dir
+    repo.dvc_dir = fs.path.join(root_dir, repo.DVC_DIR)
+    repo._reset()  # pylint: disable=protected-access
 
     if cwd_parts:
         cwd = repo.fs.path.join("/", *cwd_parts)
@@ -149,9 +154,12 @@ def switch(repo: "Repo", rev: str) -> Iterator[str]:
 
     saved_fs = repo.fs
     saved_root = repo.root_dir
+    saved_dvc_dir = repo.dvc_dir
     try:
         _switch_fs(repo, rev, repo_root_parts, cwd_parts)
         yield rev
     finally:
         repo.fs = saved_fs
         repo.root_dir = saved_root
+        repo.dvc_dir = saved_dvc_dir
+        repo._reset()  # pylint: disable=protected-access

dvc/repo/experiments/collect.py

@@ -1,6 +1,7 @@
 import itertools
 import logging
 import os
+from dataclasses import fields
 from datetime import datetime
 from typing import (
     TYPE_CHECKING,
@@ -152,7 +153,23 @@ def collect_queued(
     """
     if not baseline_revs:
         return {}
-    return repo.experiments.celery_queue.collect_queued_data(baseline_revs, **kwargs)
+    queued_data = {}
+    for rev, ranges in repo.experiments.celery_queue.collect_queued_data(
+        baseline_revs, **kwargs
+    ).items():
+        for exp_range in ranges:
+            for exp_state in exp_range.revs:
+                if exp_state.data:
+                    attrs = [f.name for f in fields(SerializableExp)]
+                    exp_state.data = SerializableExp(
+                        **{
+                            attr: getattr(exp_state.data, attr)
+                            for attr in attrs
+                            if attr != "metrics"
+                        }
+                    )
+        queued_data[rev] = ranges
+    return queued_data
 
 
 def collect_active(

dvc/repo/fetch.py

@@ -21,19 +21,25 @@ def _collect_indexes(  # noqa: PLR0913
     revs=None,
     max_size=None,
     types=None,
+    config=None,
 ):
     indexes = {}
     collection_exc = None
+
+    config = config or {}
+    if remote:
+        core = config.get("core") or {}
+        core["remote"] = remote
+        config["core"] = core
+
     for rev in repo.brancher(
         revs=revs,
         all_branches=all_branches,
         all_tags=all_tags,
         all_commits=all_commits,
     ):
-        saved_remote = repo.config["core"].get("remote")
         try:
-            if remote:
-                repo.config["core"]["remote"] = remote
+            repo.config.update(config)
 
             idx = repo.index.targets_view(
                 targets,
@@ -46,9 +52,6 @@ def _collect_indexes(  # noqa: PLR0913
         except Exception as exc:  # pylint: disable=broad-except
             collection_exc = exc
             logger.exception("failed to collect '%s'", rev or "workspace")
-        finally:
-            if remote:
-                repo.config["core"]["remote"] = saved_remote
 
     if not indexes and collection_exc:
         raise collection_exc
@@ -71,6 +74,7 @@ def fetch(  # noqa: C901, PLR0913
     revs=None,
     max_size=None,
     types=None,
+    config=None,
 ) -> int:
     """Download data items from a cloud and imported repositories
 
@@ -114,6 +118,7 @@ def fetch(  # noqa: C901, PLR0913
         revs=revs,
         max_size=max_size,
         types=types,
+        config=config,
     )
 
     cache_key = ("fetch", tokenize(sorted(indexes.keys())))

dvc/repo/get.py

@@ -22,7 +22,6 @@ class GetDVCFileError(DvcException):
 def get(url, path, out=None, rev=None, jobs=None, force=False, config=None):
     from dvc.config import Config
     from dvc.dvcfile import is_valid_filename
-    from dvc.fs.callbacks import Callback
     from dvc.repo import Repo
 
     out = resolve_output(path, out, force=force)
@@ -40,6 +39,7 @@ def get(url, path, out=None, rev=None, jobs=None, force=False, config=None):
         uninitialized=True,
         config=config,
     ) as repo:
+        from dvc.fs import download
         from dvc.fs.data import DataFileSystem
 
         fs: Union[DataFileSystem, "DVCFileSystem"]
@@ -49,14 +49,9 @@ def get(url, path, out=None, rev=None, jobs=None, force=False, config=None):
         else:
             fs = repo.dvcfs
             fs_path = fs.from_os_path(path)
-
-        with Callback.as_tqdm_callback(
-            desc=f"Downloading {fs.path.name(path)}",
-            unit="files",
-        ) as cb:
-            fs.get(
-                fs_path,
-                os.path.abspath(out),
-                batch_size=jobs,
-                callback=cb,
-            )
+        download(
+            fs,
+            fs_path,
+            os.path.abspath(out),
+            jobs=jobs,
+        )

dvc/repo/get_url.py

@@ -14,4 +14,4 @@ def get_url(url, out=None, *, config=None, jobs=None, force=False):
     src_fs, src_path = parse_external_url(url, config)
     if not src_fs.exists(src_path):
         raise URLMissingError(url)
-    download(src_fs, src_path, out, jobs=jobs)
+    download(src_fs, src_path, out.fs_path, jobs=jobs)

tests/func/repro/test_repro.py

@@ -10,7 +10,7 @@ from funcy import lsplit
 from dvc.cli import main
 from dvc.dvcfile import LOCK_FILE, PROJECT_FILE
 from dvc.exceptions import CyclicGraphError, ReproductionError
-from dvc.fs import LocalFileSystem, system
+from dvc.fs import system
 from dvc.output import Output
 from dvc.stage import PipelineStage, Stage
 from dvc.stage.cache import RunCacheNotSupported
@@ -624,12 +624,14 @@ class TestReproAlreadyCached:
         assert stage.outs[0].hash_info != saved_stage.outs[0].hash_info
 
     def test_force_import(self, mocker, tmp_dir, dvc):
+        from dvc.dependency import base
+
         tmp_dir.dvc_gen("foo", "foo")
 
         ret = main(["import-url", "foo", "bar"])
         assert ret == 0
 
-        spy_get = mocker.spy(LocalFileSystem, "get")
+        spy_get = mocker.spy(base, "fs_download")
         spy_checkout = mocker.spy(Output, "checkout")
 
         assert main(["unfreeze", "bar.dvc"]) == 0

tests/func/test_data_cloud.py

@@ -203,7 +203,8 @@ def test_verify_hashes(tmp_dir, scm, dvc, mocker, tmp_path_factory, local_remote
     # Removing cache will invalidate existing state entries
     dvc.cache.local.clear()
 
-    dvc.config["remote"]["upstream"]["verify"] = True
+    with dvc.config.edit() as conf:
+        conf["remote"]["upstream"]["verify"] = True
 
     dvc.pull()
     assert hash_spy.call_count == 10

tests/func/test_gc.py

@@ -196,11 +196,7 @@ def test_gc_cloud_positive(tmp_dir, dvc, tmp_path_factory, local_remote):
         assert main(["gc", "-vf", flag]) == 0
 
 
-def test_gc_cloud_remove_order(tmp_dir, scm, dvc, tmp_path_factory, mocker):
-    storage = os.fspath(tmp_path_factory.mktemp("test_remote_base"))
-    dvc.config["remote"]["local_remote"] = {"url": storage}
-    dvc.config["core"]["remote"] = "local_remote"
-
+def test_gc_cloud_remove_order(tmp_dir, scm, dvc, mocker, local_remote):
     (standalone, dir1, dir2) = tmp_dir.dvc_gen(
         {
             "file1": "standalone",
@@ -321,11 +317,7 @@ def test_date(tmp_dir, scm, dvc):
     )  # "modified, again"
 
 
-def test_gc_not_in_remote(tmp_dir, scm, dvc, tmp_path_factory, mocker):
-    storage = os.fspath(tmp_path_factory.mktemp("test_remote_base"))
-    dvc.config["remote"]["local_remote"] = {"url": storage}
-    dvc.config["core"]["remote"] = "local_remote"
-
+def test_gc_not_in_remote(tmp_dir, scm, dvc, mocker, local_remote):
     (standalone, dir1, dir2) = tmp_dir.dvc_gen(
         {
             "file1": "standalone",
@@ -358,12 +350,9 @@ def test_gc_not_in_remote(tmp_dir, scm, dvc, tmp_path_factory, mocker):
     )
 
 
-def test_gc_not_in_remote_remote_arg(tmp_dir, scm, dvc, tmp_path_factory, mocker):
-    storage = os.fspath(tmp_path_factory.mktemp("test_remote_base"))
-    dvc.config["remote"]["local_remote"] = {"url": storage}
-    dvc.config["core"]["remote"] = "local_remote"
-    other_storage = os.fspath(tmp_path_factory.mktemp("test_remote_other"))
-    dvc.config["remote"]["other_remote"] = {"url": other_storage}
+def test_gc_not_in_remote_remote_arg(tmp_dir, scm, dvc, mocker, make_remote):
+    make_remote("local_remote", typ="local")
+    make_remote("other_remote", typ="local", default=False)
 
     tmp_dir.dvc_gen(
         {
@@ -385,15 +374,9 @@ def test_gc_not_in_remote_remote_arg(tmp_dir, scm, dvc, tmp_path_factory, mocker
     assert len(mocked_remove.mock_calls) == 3
 
 
-def test_gc_not_in_remote_with_remote_field(
-    tmp_dir, scm, dvc, tmp_path_factory, mocker
-):
-    storage = os.fspath(tmp_path_factory.mktemp("test_remote_base"))
-    dvc.config["remote"]["local_remote"] = {"url": storage}
-    dvc.config["core"]["remote"] = "local_remote"
-
-    other_storage = os.fspath(tmp_path_factory.mktemp("test_remote_other"))
-    dvc.config["remote"]["other_remote"] = {"url": other_storage}
+def test_gc_not_in_remote_with_remote_field(tmp_dir, scm, dvc, mocker, make_remote):
+    make_remote("local_remote", typ="local")
+    make_remote("other_remote", typ="local", default=False)
 
     text = textwrap.dedent(
         """\
@@ -420,12 +403,9 @@ def test_gc_not_in_remote_cloud(tmp_dir, scm, dvc):
         dvc.gc(workspace=True, not_in_remote=True, cloud=True)
 
 
-def test_gc_cloud_remote_field(tmp_dir, scm, dvc, tmp_path_factory, mocker):
-    storage = os.fspath(tmp_path_factory.mktemp("test_remote_base"))
-    dvc.config["remote"]["local_remote"] = {"url": storage}
-    dvc.config["core"]["remote"] = "local_remote"
-    other_storage = os.fspath(tmp_path_factory.mktemp("test_remote_other"))
-    dvc.config["remote"]["other_remote"] = {"url": other_storage}
+def test_gc_cloud_remote_field(tmp_dir, scm, dvc, mocker, make_remote):
+    make_remote("local_remote", typ="local")
+    make_remote("other_remote", typ="local", default=False)
 
     text = textwrap.dedent(
         """\

tests/unit/fs/test_fs.py

@@ -1,5 +1,3 @@
-from dataclasses import dataclass
-
 import pytest
 from dvc_http import HTTPFileSystem, HTTPSFileSystem
 from dvc_s3 import S3FileSystem
@@ -63,14 +61,8 @@ def test_remote_url():
     }
 
 
-@dataclass
-class FakeRepo:
-    config: dict
-
-
 def test_get_cloud_fs():
-    repo = FakeRepo({})
-    cls, config, path = get_cloud_fs(repo, url="ssh://example.com:/dir/path")
+    cls, config, path = get_cloud_fs({}, url="ssh://example.com:/dir/path")
     assert cls is SSHFileSystem
     assert config == {"host": "example.com", "verify": False}
     assert path == "/dir/path"

tests/unit/fs/test_tree.py

@@ -13,11 +13,11 @@ def test_get_cloud_fs(tmp_dir, dvc):
     first = f"{base}/first"
     second = f"{first}/second"
 
-    _, _, path = get_cloud_fs(dvc, name="base")
+    _, _, path = get_cloud_fs(dvc.config, name="base")
     assert path == base
-    _, _, path = get_cloud_fs(dvc, name="first")
+    _, _, path = get_cloud_fs(dvc.config, name="first")
     assert path == first
-    _, _, path = get_cloud_fs(dvc, name="second")
+    _, _, path = get_cloud_fs(dvc.config, name="second")
     assert path == second
 
 
@@ -34,9 +34,9 @@ def test_get_cloud_fs_validate(tmp_dir, dvc):
         default=False,
     )
 
-    assert get_cloud_fs(dvc, name="base")[1]["host"] == "example.com"
-    assert get_cloud_fs(dvc, name="first")[1]["host"] == "example.com"
-    assert get_cloud_fs(dvc, name="first")[1]["type"] == ["symlink"]
+    assert get_cloud_fs(dvc.config, name="base")[1]["host"] == "example.com"
+    assert get_cloud_fs(dvc.config, name="first")[1]["host"] == "example.com"
+    assert get_cloud_fs(dvc.config, name="first")[1]["type"] == ["symlink"]
 
     with pytest.raises(ConfigError):
-        get_cloud_fs(dvc, name="second")
+        get_cloud_fs(dvc.config, name="second")

tests/unit/remote/test_remote.py

@@ -12,7 +12,7 @@ def test_remote_with_hash_jobs(dvc):
     }
     dvc.config["core"]["checksum_jobs"] = 200
 
-    cls, config, _ = get_cloud_fs(dvc, name="with_hash_jobs")
+    cls, config, _ = get_cloud_fs(dvc.config, name="with_hash_jobs")
     fs = cls(**config)
     assert fs.hash_jobs == 100
 
@@ -23,7 +23,7 @@ def test_remote_with_jobs(dvc):
         "jobs": 100,
     }
 
-    cls, config, _ = get_cloud_fs(dvc, name="with_jobs")
+    cls, config, _ = get_cloud_fs(dvc.config, name="with_jobs")
     fs = cls(**config)
     assert fs.jobs == 100
 
@@ -32,7 +32,7 @@ def test_remote_without_hash_jobs(dvc):
     dvc.config["remote"]["without_hash_jobs"] = {"url": "s3://bucket/name"}
     dvc.config["core"]["checksum_jobs"] = 200
 
-    cls, config, _ = get_cloud_fs(dvc, name="without_hash_jobs")
+    cls, config, _ = get_cloud_fs(dvc.config, name="without_hash_jobs")
     fs = cls(**config)
     assert fs.hash_jobs == 200
 
@@ -40,7 +40,7 @@ def test_remote_without_hash_jobs(dvc):
 def test_remote_without_hash_jobs_default(dvc):
     dvc.config["remote"]["without_hash_jobs"] = {"url": "s3://bucket/name"}
 
-    cls, config, _ = get_cloud_fs(dvc, name="without_hash_jobs")
+    cls, config, _ = get_cloud_fs(dvc.config, name="without_hash_jobs")
     fs = cls(**config)
     assert fs.hash_jobs == fs.HASH_JOBS
 

tests/unit/remote/test_webdav.py

@@ -145,14 +145,14 @@ def test_remote_with_jobs(dvc, base_url, fs_cls):
     remote_config = {"url": base_url}
 
     dvc.config["remote"]["dav"] = remote_config
-    cls, config, _ = get_cloud_fs(dvc, name="dav")
+    cls, config, _ = get_cloud_fs(dvc.config, name="dav")
     assert config["user"] == user
     assert f"{scheme}://{user}@example.com" in config["host"]
     assert cls is fs_cls
 
     # config from remote takes priority
     remote_config.update({"user": "admin"})
-    cls, config, _ = get_cloud_fs(dvc, name="dav")
+    cls, config, _ = get_cloud_fs(dvc.config, name="dav")
     assert config["user"] == "admin"
     assert f"{scheme}://{user}@example.com" in config["host"]
     assert cls is fs_cls