get: use Repo.open instead of external_repo

This should use `Repo.open` just like `dvc import`, to benefit from
handling local repos without cloning and redundant downloaads.
This commit is contained in:
Ruslan Kuprieiev
2023-05-07 04:14:25 +03:00
parent eb592e7e33
commit 01416df5a4
3 changed files with 29 additions and 63 deletions
+25 -51
View File
@@ -4,7 +4,6 @@ from typing import TYPE_CHECKING, Union
from dvc.exceptions import DvcException
from dvc.utils import resolve_output
from dvc.utils.fs import remove
if TYPE_CHECKING:
from dvc.fs.dvc import DVCFileSystem
@@ -21,63 +20,38 @@ class GetDVCFileError(DvcException):
def get(url, path, out=None, rev=None, jobs=None, force=False):
import shortuuid
from dvc.dvcfile import is_valid_filename
from dvc.fs.callbacks import Callback
from dvc.repo.open_repo import external_repo
from dvc.repo import Repo
out = resolve_output(path, out, force=force)
if is_valid_filename(out):
raise GetDVCFileError()
# Creating a directory right beside the output to make sure that they
# are on the same filesystem, so we could take the advantage of
# reflink and/or hardlink. Not using tempfile.TemporaryDirectory
# because it will create a symlink to tmpfs, which defeats the purpose
# and won't work with reflink/hardlink.
dpath = os.path.dirname(os.path.abspath(out))
tmp_dir = os.path.join(dpath, "." + str(shortuuid.uuid()))
with Repo.open(
url=url,
rev=rev,
subrepos=True,
uninitialized=True,
) as repo:
from dvc.fs.data import DataFileSystem
# Try any links possible to avoid data duplication.
#
# Not using symlink, because we need to remove cache after we
# are done, and to make that work we would have to copy data
# over anyway before removing the cache, so we might just copy
# it right away.
#
# Also, we can't use theoretical "move" link type here, because
# the same cache file might be used a few times in a directory.
cache_types = ["reflink", "hardlink", "copy"]
try:
with external_repo(
url=url,
rev=rev,
subrepos=True,
uninitialized=True,
cache_dir=tmp_dir,
cache_types=cache_types,
) as repo:
from dvc.fs.data import DataFileSystem
fs: Union[DataFileSystem, "DVCFileSystem"]
if os.path.isabs(path):
fs = DataFileSystem(index=repo.index.data["local"])
fs_path = fs.from_os_path(path)
else:
fs = repo.dvcfs
fs_path = fs.from_os_path(path)
fs: Union[DataFileSystem, "DVCFileSystem"]
if os.path.isabs(path):
fs = DataFileSystem(index=repo.index.data["local"])
fs_path = fs.from_os_path(path)
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,
)
finally:
remove(tmp_dir)
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,
)
+3 -10
View File
@@ -3,7 +3,7 @@ import os
import tempfile
import threading
from contextlib import contextmanager
from typing import TYPE_CHECKING, Collection, Dict, Iterator, Optional, Tuple
from typing import TYPE_CHECKING, Dict, Iterator, Optional, Tuple
from funcy import retry, wrap_with
@@ -14,7 +14,6 @@ from dvc.utils import relpath
if TYPE_CHECKING:
from dvc.scm import Git
from dvc.types import StrPath
logger = logging.getLogger(__name__)
@@ -25,8 +24,6 @@ def external_repo(
url,
rev: Optional[str] = None,
for_write: bool = False,
cache_dir: Optional["StrPath"] = None,
cache_types: Optional[Collection[str]] = None,
**kwargs,
) -> Iterator["Repo"]:
logger.debug("Creating external repo %s@%s", url, rev)
@@ -36,12 +33,8 @@ def external_repo(
# the tip of the default branch
rev = rev or "refs/remotes/origin/HEAD"
cache_config = {
"cache": {"dir": cache_dir or _get_cache_dir(url), "type": cache_types}
}
config = _get_remote_config(url) if os.path.isdir(url) else {}
config.update(cache_config)
config.update({"cache": {"dir": _get_cache_dir(url)}})
config.update(kwargs.pop("config", None) or {})
main_root = "/"
@@ -54,7 +47,7 @@ def external_repo(
root_dir=path,
url=url,
config=config,
repo_factory=erepo_factory(url, main_root, cache_config),
repo_factory=erepo_factory(url, main_root, {"cache": config["cache"]}),
rev=rev,
**kwargs,
)
+1 -2
View File
@@ -65,8 +65,7 @@ def test_subrepo_is_constructed_properly(
str(tmp_dir),
subrepos=True,
uninitialized=True,
cache_dir=str(cache_dir),
cache_types=["symlink"],
config={"cache": {"dir": str(cache_dir), "type": ["symlink"]}},
) as repo:
spy = mocker.spy(repo.dvcfs.fs, "repo_factory")