Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-07 07:18:27 +01:00
Code Issues Packages Projects Releases Wiki Activity

dvcfs: base on fsspec

Browse Source
This commit is contained in:
Ruslan Kuprieiev
2022-03-31 17:55:18 +03:00
parent 0470e164cc
commit d600eeca99
6 changed files with 68 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/api.py
+4
View File
@@ -28,6 +28,10 @@ def get_url(path, repo=None, rev=None, remote=None):
cloud = info["repo"].cloud
dvc_path = _repo.fs.path.relpath(fs_path, info["repo"].root_dir)
if not os.path.isabs(path):
dvc_path = dvc_path.replace("\\", "/")
md5 = info["repo"].dvcfs.info(dvc_path)["md5"]
return cloud.get_url_for(remote, checksum=md5)
dvc/data/tree.py
+5 -1
View File
@@ -19,6 +19,10 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
class TreeError(Exception):
pass
def _try_load(
odbs: Iterable["ObjectDB"],
hash_info: "HashInfo",
@@ -204,7 +208,7 @@ class Tree(HashFile):
meta, hash_info = self._trie.get(prefix, (None, None))
if hash_info and hash_info.isdir and meta and not meta.obj:
raise FileNotFoundError
raise TreeError
ret = []
dvc/dependency/repo.py
+2 -2
View File
@@ -99,7 +99,7 @@ class RepoDependency(Dependency):
) -> Tuple[Dict[Optional["ObjectDB"], Set["HashInfo"]], "HashFile"]:
from dvc.config import NoRemoteError
from dvc.data.stage import stage
from dvc.data.tree import Tree
from dvc.data.tree import Tree, TreeError
from dvc.exceptions import NoOutputOrStageError, PathMissingError
local_odb = self.repo.odb.local
@@ -136,7 +136,7 @@ class RepoDependency(Dependency):
repo.repo_fs,
local_odb.fs.PARAM_CHECKSUM,
)
except FileNotFoundError as exc:
except (FileNotFoundError, TreeError) as exc:
raise PathMissingError(
self.def_path, self.def_repo[self.PARAM_URL]
) from exc
dvc/fs/dvc.py
+34 -71
View File
@@ -1,9 +1,13 @@
import logging
import os
import threading
import typing
from fsspec import AbstractFileSystem
from funcy import cached_property, wrap_prop
from ._callback import DEFAULT_CALLBACK
from .base import FileSystem
from .fsspec_wrapper import FSSpecWrapper
if typing.TYPE_CHECKING:
from dvc.types import AnyPath
@@ -11,18 +15,13 @@ if typing.TYPE_CHECKING:
logger = logging.getLogger(__name__)
class DvcFileSystem(FileSystem): # pylint:disable=abstract-method
class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
"""DVC repo fs.
Args:
repo: DVC repo.
"""
sep = os.sep
scheme = "local"
PARAM_CHECKSUM = "md5"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.repo = kwargs["repo"]
@@ -43,8 +42,8 @@ class DvcFileSystem(FileSystem): # pylint:disable=abstract-method
return (cls.scheme, *fs.path.parts(fs_path))
fs_key = "repo"
key = self.path.parts(path)
if key == (".",) or key == ("",):
key = path.split(self.sep)
if key == ["."] or key == [""]:
key = ()
return (fs_key, *key)
@@ -74,29 +73,10 @@ class DvcFileSystem(FileSystem): # pylint:disable=abstract-method
def open( # type: ignore
self, path: str, mode="r", encoding=None, **kwargs
): # pylint: disable=arguments-renamed
): # pylint: disable=arguments-renamed, arguments-differ
fs, fspath = self._get_fs_path(path, **kwargs)
return fs.open(fspath, mode=mode, encoding=encoding)
def exists(self, path): # pylint: disable=arguments-renamed
try:
self.info(path)
return True
except FileNotFoundError:
return False
def isdir(self, path): # pylint: disable=arguments-renamed
try:
return self.info(path)["type"] == "directory"
except FileNotFoundError:
return False
def isfile(self, path): # pylint: disable=arguments-renamed
try:
return self.info(path)["type"] == "file"
except FileNotFoundError:
return False
def ls(self, path, detail=True, **kwargs):
info = self.info(path)
if info["type"] != "directory":
@@ -116,42 +96,6 @@ class DvcFileSystem(FileSystem): # pylint:disable=abstract-method
return [self.info(epath) for epath in entries]
def _walk(self, root, topdown=True, **kwargs):
dirs = []
files = []
for entry in self.ls(root, detail=True):
name = self.path.name(entry["name"])
if entry["type"] == "directory":
dirs.append(name)
else:
files.append(name)
assert topdown
dirs = list(dirs)
yield root, dirs, files
for dname in dirs:
yield from self._walk(self.path.join(root, dname))
def walk(self, top, topdown=True, **kwargs):
assert topdown
try:
info = self.info(top)
except FileNotFoundError:
return
if info["type"] != "directory":
return
yield from self._walk(top, topdown=topdown, **kwargs)
def find(self, path, prefix=None):
for root, _, files in self.walk(path):
for fname in files:
# NOTE: os.path.join is ~5.5 times slower
yield f"{root}{os.sep}{fname}"
def isdvc(self, path, recursive=False, strict=True):
try:
info = self.info(path)
@@ -161,7 +105,7 @@ class DvcFileSystem(FileSystem): # pylint:disable=abstract-method
recurse = recursive or not strict
return bool(info.get("outs") if recurse else info.get("isout"))
def info(self, path):
def info(self, path, **kwargs):
from dvc.data.meta import Meta
key = self._get_key(path)
@@ -209,12 +153,10 @@ class DvcFileSystem(FileSystem): # pylint:disable=abstract-method
ret["type"] = "directory"
return ret
def get_file(
self, from_info, to_file, callback=DEFAULT_CALLBACK, **kwargs
):
fs, path = self._get_fs_path(from_info)
def get_file(self, rpath, lpath, callback=DEFAULT_CALLBACK, **kwargs):
fs, path = self._get_fs_path(rpath)
fs.get_file( # pylint: disable=protected-access
path, to_file, callback=callback, **kwargs
path, lpath, callback=callback, **kwargs
)
def checksum(self, path):
@@ -223,3 +165,24 @@ class DvcFileSystem(FileSystem): # pylint:disable=abstract-method
if md5:
return md5
raise NotImplementedError
class DvcFileSystem(FSSpecWrapper):
scheme = "local"
PARAM_CHECKSUM = "md5"
def _prepare_credentials(self, **config):
return config
@wrap_prop(threading.Lock())
@cached_property
def fs(self):
return _DvcFileSystem(**self.fs_args)
def isdvc(self, path, **kwargs):
return self.fs.isdvc(path, **kwargs)
@property
def repo(self):
return self.fs.repo
dvc/fs/repo.py
+2
View File
@@ -228,6 +228,8 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
if path.startswith(repo.root_dir):
dvc_path = path[len(repo.root_dir) + 1 :]
dvc_path = dvc_path.replace("\\", "/")
else:
dvc_path = path
tests/unit/fs/test_dvc.py
+21 -21
View File
@@ -1,4 +1,4 @@
import os
import posixpath
import shutil
import pytest
@@ -16,12 +16,12 @@ from dvc.utils.fs import remove
("", ("repo",)),
(".", ("repo",)),
("foo", ("repo", "foo")),
(os.path.join("dir", "foo"), ("repo", "dir", "foo")),
("dir/foo", ("repo", "dir", "foo")),
],
)
def test_get_key(tmp_dir, dvc, path, key):
fs = DvcFileSystem(repo=dvc)
assert fs._get_key(path) == key
assert fs.fs._get_key(path) == key
def test_exists(tmp_dir, dvc):
@@ -143,19 +143,19 @@ def test_walk(tmp_dir, dvc):
fs = DvcFileSystem(repo=dvc)
expected = [
os.path.join("dir", "subdir1"),
os.path.join("dir", "subdir2"),
os.path.join("dir", "subdir1", "foo1"),
os.path.join("dir", "subdir1", "bar1"),
os.path.join("dir", "subdir2", "foo2"),
os.path.join("dir", "foo"),
os.path.join("dir", "bar"),
"dir/subdir1",
"dir/subdir2",
"dir/subdir1/foo1",
"dir/subdir1/bar1",
"dir/subdir2/foo2",
"dir/foo",
"dir/bar",
]
actual = []
for root, dirs, files in fs.walk("dir"):
for entry in dirs + files:
actual.append(os.path.join(root, entry))
actual.append(posixpath.join(root, entry))
assert set(actual) == set(expected)
assert len(actual) == len(expected)
@@ -177,19 +177,19 @@ def test_walk_dir(tmp_dir, dvc):
fs = DvcFileSystem(repo=dvc)
expected = [
os.path.join("dir", "subdir1"),
os.path.join("dir", "subdir2"),
os.path.join("dir", "subdir1", "foo1"),
os.path.join("dir", "subdir1", "bar1"),
os.path.join("dir", "subdir2", "foo2"),
os.path.join("dir", "foo"),
os.path.join("dir", "bar"),
"dir/subdir1",
"dir/subdir2",
"dir/subdir1/foo1",
"dir/subdir1/bar1",
"dir/subdir2/foo2",
"dir/foo",
"dir/bar",
]
actual = []
for root, dirs, files in fs.walk("dir"):
for entry in dirs + files:
actual.append(os.path.join(root, entry))
actual.append(posixpath.join(root, entry))
assert set(actual) == set(expected)
assert len(actual) == len(expected)
@@ -241,13 +241,13 @@ def test_get_hash_granular(tmp_dir, dvc):
{"dir": {"foo": "foo", "bar": "bar", "subdir": {"data": "data"}}}
)
fs = DvcFileSystem(repo=dvc)
subdir = os.path.join("dir", "subdir")
subdir = "dir/subdir"
assert fs.info(subdir).get("md5") is None
_, _, obj = stage(dvc.odb.local, subdir, fs, "md5", dry_run=True)
assert obj.hash_info == HashInfo(
"md5", "af314506f1622d107e0ed3f14ec1a3b5.dir"
)
data = os.path.join(subdir, "data")
data = posixpath.join(subdir, "data")
assert fs.info(data)["md5"] == "8d777f385d3dfec8815d20f7496026dc"
_, _, obj = stage(dvc.odb.local, data, fs, "md5", dry_run=True)
assert obj.hash_info == HashInfo("md5", "8d777f385d3dfec8815d20f7496026dc")