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

datafs: use dvc-data's datafs

Browse Source
This commit is contained in:
Ruslan Kuprieiev
2022-08-02 03:13:06 +03:00
parent ed62ddb373
commit 6a7b55fdf7
10 changed files with 87 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/fs/data.py
+8 -173
View File
@@ -1,183 +1,16 @@
import logging
import os
import threading
import typing
from fsspec import AbstractFileSystem
from funcy import cached_property, wrap_prop
from dvc.utils import as_posix
from dvc_data.fs import DataFileSystem as _DataFileSystem
from dvc_objects.fs.base import FileSystem
from dvc_objects.fs.callbacks import DEFAULT_CALLBACK
if typing.TYPE_CHECKING:
from dvc.types import AnyPath
logger = logging.getLogger(__name__)
class _DataFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
"""DVC repo fs.
Args:
repo: DVC repo.
"""
root_marker = "/"
def __init__(self, workspace=None, **kwargs):
super().__init__(**kwargs)
self.repo = kwargs["repo"]
self.workspace = workspace or "repo"
@cached_property
def path(self):
from . import Path
def _getcwd():
return self.root_marker
return Path(self.sep, getcwd=_getcwd)
@property
def config(self):
raise NotImplementedError
def _get_key(self, path):
if self.workspace != "repo":
from . import get_cloud_fs
cls, kwargs, fs_path = get_cloud_fs(None, url=path)
fs = cls(**kwargs)
return (self.workspace, *fs.path.parts(fs_path))
path = self.path.abspath(path)
if path == self.root_marker:
return (self.workspace,)
key = self.path.relparts(path, self.root_marker)
if key == (".") or key == (""):
key = ()
return (self.workspace, *key)
def _get_fs_path(self, path: "AnyPath", remote=None):
from dvc.config import NoRemoteError
info = self.info(path)
if info["type"] == "directory":
raise IsADirectoryError
value = info.get("md5")
if not value:
raise FileNotFoundError
cache_path = self.repo.odb.local.oid_to_path(value)
if self.repo.odb.local.fs.exists(cache_path):
return self.repo.odb.local.fs, cache_path
try:
remote_odb = self.repo.cloud.get_remote_odb(remote)
except NoRemoteError as exc:
raise FileNotFoundError from exc
remote_fs_path = remote_odb.oid_to_path(value)
return remote_odb.fs, remote_fs_path
def open( # type: ignore
self, path: str, mode="r", encoding=None, **kwargs
): # pylint: disable=arguments-renamed, arguments-differ
fs, fspath = self._get_fs_path(path, **kwargs)
return fs.open(fspath, mode=mode, encoding=encoding)
def ls(self, path, detail=True, **kwargs):
info = self.info(path)
if info["type"] != "directory":
return [info] if detail else [path]
root_key = self._get_key(path)
try:
entries = [
self.sep.join((path, name)) if path else name
for name in self.repo.index.tree.ls(prefix=root_key)
]
except KeyError as exc:
raise FileNotFoundError from exc
if not detail:
return entries
return [self.info(epath) for epath in entries]
def isdvc(self, path, recursive=False, strict=True):
try:
info = self.info(path)
except FileNotFoundError:
return False
recurse = recursive or not strict
return bool(info.get("outs") if recurse else info.get("isout"))
def info(self, path, **kwargs):
from dvc_data.hashfile.meta import Meta
key = self._get_key(path)
try:
outs = list(self.repo.index.tree.iteritems(key)) # noqa: B301
except KeyError as exc:
raise FileNotFoundError from exc
ret = {
"type": "file",
"size": 0,
"isexec": False,
"isdvc": False,
"outs": outs,
"name": path,
}
if len(outs) > 1 and outs[0][0] != key:
shortest = self.repo.index.tree.shortest_prefix(key)
if shortest:
assert shortest[1][1].isdir
if len(shortest[0]) <= len(key):
ret["isdvc"] = True
ret["type"] = "directory"
return ret
item_key, (meta, hash_info) = outs[0]
meta = meta or Meta()
if key != item_key:
assert item_key[: len(key)] == key
ret["type"] = "directory"
return ret
ret["size"] = meta.size
ret["isexec"] = meta.isexec
ret[hash_info.name] = hash_info.value
ret["isdvc"] = True
ret["isout"] = True
ret["meta"] = meta
if hash_info and hash_info.isdir:
ret["type"] = "directory"
return ret
def get_file( # pylint: disable=arguments-differ
self, rpath, lpath, callback=DEFAULT_CALLBACK, **kwargs
):
fs, path = self._get_fs_path(rpath)
fs.get_file(path, lpath, callback=callback, **kwargs)
def checksum(self, path):
info = self.info(path)
md5 = info.get("md5")
if md5:
return md5
raise NotImplementedError
class DataFileSystem(FileSystem):
protocol = "local"
@@ -194,6 +27,8 @@ class DataFileSystem(FileSystem):
def isdvc(self, path, **kwargs):
return self.fs.isdvc(path, **kwargs)
@property
def repo(self):
return self.fs.repo
def from_os_path(self, path):
if os.path.isabs(path):
path = os.path.splitdrive(path)[1]
return as_posix(path)
dvc/fs/dvc.py
+12 -11
View File
@@ -135,7 +135,7 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
"""Keep a datafs instance of each repo."""
if hasattr(repo, "dvc_dir"):
self._datafss[key] = DataFileSystem(repo=repo)
self._datafss[key] = DataFileSystem(index=repo.index.data["repo"])
def _get_key(self, path):
parts = self.repo.fs.path.relparts(path, self.repo.root_dir)
@@ -252,7 +252,9 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
scm=self.repo.scm,
repo_factory=self.repo_factory,
)
self._datafss[key] = DataFileSystem(repo=repo)
self._datafss[key] = DataFileSystem(
index=repo.index.data["repo"]
)
self._subrepos_trie[key] = repo
def _is_dvc_repo(self, dir_path):
@@ -268,6 +270,7 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
def _get_fs_pair(
self, path
) -> Tuple[
Optional["Repo"],
Optional[FileSystem],
Optional[str],
Optional[DataFileSystem],
@@ -299,7 +302,7 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
else:
dvc_path = None
return fs, fs_path, dvc_fs, dvc_path
return repo, fs, fs_path, dvc_fs, dvc_path
def open(
self, path, mode="r", encoding="utf-8", **kwargs
@@ -307,7 +310,7 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
if "b" in mode:
encoding = None
fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
_, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
try:
return fs.open(fs_path, mode=mode, encoding=encoding)
except FileNotFoundError:
@@ -317,15 +320,14 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
return dvc_fs.open(dvc_path, mode=mode, encoding=encoding, **kwargs)
def isdvc(self, path, **kwargs):
_, _, dvc_fs, dvc_path = self._get_fs_pair(path)
_, _, _, dvc_fs, dvc_path = self._get_fs_pair(path)
return dvc_fs is not None and dvc_fs.isdvc(dvc_path, **kwargs)
def ls( # pylint: disable=arguments-differ
self, path, detail=True, dvc_only=False, **kwargs
):
fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
repo, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
repo = dvc_fs.repo if dvc_fs else self.repo
dvcignore = repo.dvcignore
ignore_subrepos = kwargs.get("ignore_subrepos", True)
@@ -378,7 +380,7 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
def get_file( # pylint: disable=arguments-differ
self, rpath, lpath, callback=DEFAULT_CALLBACK, **kwargs
):
fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(rpath)
_, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(rpath)
if fs:
try:
@@ -390,9 +392,8 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
dvc_fs.get_file(dvc_path, lpath, callback=callback, **kwargs)
def info(self, path, **kwargs):
fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
repo, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
repo = dvc_fs.repo if dvc_fs else self.repo
dvcignore = repo.dvcignore
ignore_subrepos = kwargs.get("ignore_subrepos", True)
@@ -434,7 +435,7 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
return info
def checksum(self, path):
fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
_, fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
try:
return fs.checksum(fs_path)
dvc/repo/__init__.py
+8 -3
View File
@@ -480,7 +480,7 @@ class Repo:
def datafs(self):
from dvc.fs.data import DataFileSystem
return DataFileSystem(repo=self)
return DataFileSystem(index=self.index.data["repo"])
@cached_property
def dvcfs(self):
@@ -501,18 +501,23 @@ class Repo:
from dvc.fs.dvc import DvcFileSystem
if os.path.isabs(path):
fs = DataFileSystem(repo=self, workspace="local")
fs = DataFileSystem(index=self.index.data["local"])
fs_path = path
else:
fs = DvcFileSystem(repo=self, subrepos=True)
fs_path = fs.from_os_path(path)
try:
if remote:
remote_odb = self.cloud.get_remote_odb(name=remote)
oid = fs.info(fs_path)["dvc_info"]["md5"]
fs = remote_odb.fs
fs_path = remote_odb.oid_to_path(oid)
with fs.open(
fs_path,
mode=mode,
encoding=encoding,
remote=remote,
) as fobj:
yield fobj
except FileNotFoundError as exc:
dvc/repo/diff.py
+1 -1
View File
@@ -190,7 +190,7 @@ def _filter_missing(dvcfs, paths):
if (
dvc_info
and info["type"] == "directory"
and not dvc_info["meta"].obj
and not dvc_info["obj"]
):
yield path
except FileNotFoundError:
dvc/repo/get.py
+2 -2
View File
@@ -54,8 +54,8 @@ def get(url, path, out=None, rev=None, jobs=None):
if os.path.isabs(path):
from dvc.fs.data import DataFileSystem
fs = DataFileSystem(repo=repo, workspace="local")
fs_path = 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)
dvc/repo/index.py
+27 -13
View File
@@ -25,8 +25,8 @@ if TYPE_CHECKING:
from dvc.repo.stage import StageLoad
from dvc.stage import Stage
from dvc.types import StrPath, TargetType
from dvc_data import Tree
from dvc_data.hashfile.hash_info import HashInfo
from dvc_data.index import DataIndex
from dvc_objects.db import ObjectDB
@@ -168,33 +168,47 @@ class Index:
return build_outs_graph(self.graph, self.outs_trie)
@cached_property
def tree(self) -> "Tree":
from dvc.config import NoRemoteError
from dvc_data import Tree
def data(self) -> "Dict[str, DataIndex]":
from collections import defaultdict
tree = Tree()
from dvc.config import NoRemoteError
from dvc_data.index import DataIndex, DataIndexEntry
by_workspace: dict = defaultdict(DataIndex)
by_workspace["repo"] = DataIndex()
by_workspace["local"] = DataIndex()
for out in self.outs:
if not out.use_cache:
continue
if out.is_in_repo:
fs_key = "repo"
workspace = "repo"
key = self.repo.fs.path.relparts(
out.fs_path, self.repo.root_dir
)
else:
fs_key = out.fs.protocol
key = out.fs.path.parts(out.fs_path)
workspace = out.fs.protocol
no_drive = out.fs.path.flavour.splitdrive(out.fs_path)[1]
key = out.fs.path.parts(no_drive)[1:]
out.meta.odb = out.odb
try:
out.meta.remote = self.repo.cloud.get_remote_odb(out.remote)
remote = self.repo.cloud.get_remote_odb(out.remote)
except NoRemoteError:
out.meta.remote = None
tree.add((fs_key,) + key, out.meta, out.hash_info)
remote = None
return tree
data_index = by_workspace[workspace]
data_index[key] = DataIndexEntry(
meta=out.meta,
obj=out.obj,
hash_info=out.hash_info,
odb=out.odb,
remote=remote,
)
return dict(by_workspace)
def used_objs(
self,
dvc/repo/ls.py
+1 -1
View File
@@ -85,7 +85,7 @@ def _ls(
ret = {}
for name, info in infos.items():
dvc_info = info.get("dvc_info", {})
if dvc_info.get("outs") or not dvc_only:
if dvc_info.get("entries") or not dvc_only:
ret[name] = {
"isout": dvc_info.get("isout", False),
"isdir": info["type"] == "directory",
setup.cfg
+1 -1
View File
@@ -69,7 +69,7 @@ install_requires =
dvc-render==0.0.9
dvc-task==0.1.0
dvclive>=0.10.0
dvc-data==0.1.6
dvc-data==0.1.8
[options.extras_require]
all =
tests/unit/fs/test_data.py
+26 -27
View File
@@ -4,7 +4,6 @@ import shutil
import pytest
import dvc_data
from dvc.config import NoRemoteError
from dvc.fs.data import DataFileSystem
from dvc.utils.fs import remove
from dvc_data.build import build
@@ -14,15 +13,15 @@ from dvc_data.hashfile.hash_info import HashInfo
@pytest.mark.parametrize(
"path, key",
[
("", ("repo",)),
(".", ("repo",)),
("/", ("repo",)),
("foo", ("repo", "foo")),
("dir/foo", ("repo", "dir", "foo")),
("", ()),
(".", ()),
("/", ()),
("foo", ("foo",)),
("dir/foo", ("dir", "foo")),
],
)
def test_get_key(tmp_dir, dvc, path, key):
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
assert fs.fs._get_key(path) == key
@@ -31,7 +30,7 @@ def test_exists(tmp_dir, dvc):
dvc.add("foo")
(tmp_dir / "foo").unlink()
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
assert fs.exists("foo")
@@ -40,7 +39,7 @@ def test_open(tmp_dir, dvc):
dvc.add("foo")
(tmp_dir / "foo").unlink()
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
with fs.open("foo", "r") as fobj:
assert fobj.read() == "foo"
@@ -49,7 +48,7 @@ def test_open_dirty_hash(tmp_dir, dvc):
tmp_dir.dvc_gen("file", "file")
(tmp_dir / "file").write_text("something")
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
with fs.open("file", "r") as fobj:
# NOTE: Unlike DvcFileSystem, DataFileSystem should not
# be affected by a dirty workspace.
@@ -61,18 +60,17 @@ def test_open_no_remote(tmp_dir, dvc):
(tmp_dir / "file").unlink()
remove(dvc.odb.local.cache_dir)
fs = DataFileSystem(repo=dvc)
with pytest.raises(FileNotFoundError) as exc_info:
fs = DataFileSystem(index=dvc.index.data["repo"])
with pytest.raises(FileNotFoundError):
with fs.open("file", "r"):
pass
assert isinstance(exc_info.value.__cause__, NoRemoteError)
def test_open_dirty_no_hash(tmp_dir, dvc):
tmp_dir.gen("file", "file")
(tmp_dir / "file.dvc").write_text("outs:\n- path: file\n")
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
# NOTE: Unlike DvcFileSystem, DataFileSystem should not
# be affected by a dirty workspace.
with pytest.raises(FileNotFoundError):
@@ -95,7 +93,7 @@ def test_open_in_history(tmp_dir, scm, dvc):
if rev == "workspace":
continue
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
with fs.open("foo", "r") as fobj:
assert fobj.read() == "foo"
@@ -103,7 +101,7 @@ def test_open_in_history(tmp_dir, scm, dvc):
def test_isdir_isfile(tmp_dir, dvc):
tmp_dir.gen({"datafile": "data", "datadir": {"foo": "foo", "bar": "bar"}})
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
assert not fs.isdir("datadir")
assert not fs.isfile("datadir")
assert not fs.isdir("datafile")
@@ -113,6 +111,7 @@ def test_isdir_isfile(tmp_dir, dvc):
shutil.rmtree(tmp_dir / "datadir")
(tmp_dir / "datafile").unlink()
fs = DataFileSystem(index=dvc.index.data["repo"])
assert fs.isdir("datadir")
assert not fs.isfile("datadir")
assert not fs.isdir("datafile")
@@ -124,7 +123,7 @@ def test_isdir_mixed(tmp_dir, dvc):
dvc.add(str(tmp_dir / "dir" / "foo"))
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
assert fs.isdir("dir")
assert not fs.isfile("dir")
@@ -142,7 +141,7 @@ def test_walk(tmp_dir, dvc):
)
dvc.add("dir", recursive=True)
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
expected = [
"dir/subdir1",
@@ -176,7 +175,7 @@ def test_walk_dir(tmp_dir, dvc):
)
dvc.add("dir")
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
expected = [
"dir/subdir1",
@@ -198,7 +197,7 @@ def test_walk_dir(tmp_dir, dvc):
def test_walk_missing(tmp_dir, dvc):
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
for _ in fs.walk("dir"):
pass
@@ -206,7 +205,7 @@ def test_walk_missing(tmp_dir, dvc):
def test_walk_not_a_dir(tmp_dir, dvc):
tmp_dir.dvc_gen("foo", "foo")
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
for _ in fs.walk("foo"):
pass
@@ -215,14 +214,14 @@ def test_walk_not_a_dir(tmp_dir, dvc):
def test_isdvc(tmp_dir, dvc):
tmp_dir.gen({"foo": "foo", "bar": "bar"})
dvc.add("foo")
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
assert fs.isdvc("foo")
assert not fs.isdvc("bar")
def test_get_hash_file(tmp_dir, dvc):
tmp_dir.dvc_gen({"foo": "foo"})
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
assert fs.info("foo")["md5"] == "acbd18db4cc2f85cedef654fccc4a4d8"
@@ -230,7 +229,7 @@ def test_get_hash_dir(tmp_dir, dvc, mocker):
tmp_dir.dvc_gen(
{"dir": {"foo": "foo", "bar": "bar", "subdir": {"data": "data"}}}
)
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
hash_file_spy = mocker.spy(dvc_data.hashfile.hash, "hash_file")
assert fs.info("dir")["md5"] == "8761c4e9acad696bee718615e23e22db.dir"
assert not hash_file_spy.called
@@ -240,7 +239,7 @@ def test_get_hash_granular(tmp_dir, dvc):
tmp_dir.dvc_gen(
{"dir": {"foo": "foo", "bar": "bar", "subdir": {"data": "data"}}}
)
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
subdir = "dir/subdir"
assert fs.info(subdir).get("md5") is None
_, _, obj = build(dvc.odb.local, subdir, fs, "md5", dry_run=True)
@@ -257,7 +256,7 @@ def test_get_hash_dirty_file(tmp_dir, dvc):
tmp_dir.dvc_gen("file", "file")
(tmp_dir / "file").write_text("something")
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
expected = "8c7dd922ad47494fc02c388e12c00eac"
assert fs.info("file").get("md5") == expected
_, _, obj = build(dvc.odb.local, "file", fs, "md5", dry_run=True)
@@ -268,7 +267,7 @@ def test_get_hash_dirty_dir(tmp_dir, dvc):
tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}})
(tmp_dir / "dir" / "baz").write_text("baz")
fs = DataFileSystem(repo=dvc)
fs = DataFileSystem(index=dvc.index.data["repo"])
expected = "5ea40360f5b4ec688df672a4db9c17d1.dir"
assert fs.info("dir").get("md5") == expected
_, _, obj = build(dvc.odb.local, "dir", fs, "md5", dry_run=True)
tests/unit/fs/test_dvc.py
+1
View File
@@ -107,6 +107,7 @@ def test_isdir_isfile(tmp_dir, dvc):
(tmp_dir / "datafile").unlink()
(tmp_dir / "subdir" / "baz").unlink()
fs = DvcFileSystem(repo=dvc)
assert fs.isdir("datadir")
assert not fs.isfile("datadir")
assert fs.isdvc("datadir")