dvcfs: use info() instead of metadata()

metadata was introduced before our migration to fsspec and no longer needed
now. This is a pre-requisite to using data index in dvcfs/repofs.

Note that a bunch of awkward flags like isdvc/isout/etc are still kept, and
will be removed in followups.
This commit is contained in:
Ruslan Kuprieiev
2022-02-03 22:45:10 +02:00
parent dc13179759
commit 8497edab93
8 changed files with 274 additions and 391 deletions
+4 -4
View File
@@ -21,13 +21,13 @@ def get_url(path, repo=None, rev=None, remote=None):
with Repo.open(repo, rev=rev, subrepos=True, uninitialized=True) as _repo:
fs_path = _repo.fs.path.join(_repo.root_dir, path)
with reraise(FileNotFoundError, PathMissingError(path, repo)):
metadata = _repo.repo_fs.metadata(fs_path)
info = _repo.repo_fs.info(fs_path)
if not metadata.is_dvc:
if not info["isdvc"]:
raise OutputNotFoundError(path, repo)
cloud = metadata.repo.cloud
md5 = metadata.repo.dvcfs.info(fs_path)["md5"]
cloud = info["repo"].cloud
md5 = info["repo"].dvcfs.info(fs_path)["md5"]
return cloud.get_url_for(remote, checksum=md5)
-73
View File
@@ -1,73 +0,0 @@
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, List
from dvc.output import Output
if TYPE_CHECKING:
from dvc.repo import Repo
@dataclass
class Metadata:
"""
Container for storing metadata for a given path, similar in spirit to
`os.stat_result`.
"""
# required field
fs_path: str
repo: "Repo"
# computed fields
is_output: bool = field(init=False, default=False) # is it an output?
# is the path part of an output?
part_of_output: bool = field(init=False, default=False)
# does the path contain outputs?
contains_outputs: bool = field(init=False, default=False)
# is the path tracked by dvc? equivalent to (is_output or part_of_output)
is_dvc: bool = field(init=False, default=False)
# does dvc has any output in that path or inside it? equiv. to bool(outs)
output_exists: bool = field(init=False)
# optional fields
isdir: bool = False # is it a directory?
is_exec: bool = False # is it an executable?
outs: List[Output] = field(default_factory=list) # list of outputs
def __post_init__(self):
self.output_exists = bool(self.outs)
if not self.output_exists:
return
# it can contain multiple outputs
if len(self.outs) > 1:
self.contains_outputs = True
else:
out = self.outs[0]
# or, the path itself could be an output
self.is_output = self.fs_path == out.fs_path
# or, a directory must have an output somewhere deep inside it
if not self.is_output:
self.contains_outputs = self.repo.fs.path.isin(
out.fs_path, self.fs_path
)
# or, the path could be a part of an output, i.e. inside of an output
self.part_of_output = not self.is_output and not self.contains_outputs
self.is_dvc = self.is_output or self.part_of_output
# if it contains outputs, it must be a dir
self.isdir = self.contains_outputs
@property
def isfile(self):
# self.isdir might require additional computation, especially if it's
# a dir that's inside a dvc-tracked dir
# since, `Metadata()` does not support being used as null (i.e. when
# path does not exist), `isfile` becomes `isdir`'s complement.
return not self.isdir
def __str__(self):
return f"Metadata: {self.fs_path}"
+49 -44
View File
@@ -6,7 +6,6 @@ from dvc.exceptions import OutputNotFoundError
from dvc.utils import relpath
from ._callback import DEFAULT_CALLBACK
from ._metadata import Metadata
from .base import FileSystem
if typing.TYPE_CHECKING:
@@ -106,38 +105,20 @@ class DvcFileSystem(FileSystem): # pylint:disable=abstract-method
def exists(self, path): # pylint: disable=arguments-renamed
try:
self.metadata(path)
self.info(path)
return True
except FileNotFoundError:
return False
def isdir(self, path): # pylint: disable=arguments-renamed
try:
meta = self.metadata(path)
return meta.isdir
return self.info(path)["type"] == "directory"
except FileNotFoundError:
return False
def check_isdir(self, path, outs):
if len(outs) != 1:
return True
out = outs[0]
if not out.is_dir_checksum:
return out.fs_path != path
if out.fs_path == path:
return True
try:
self._get_granular_hash(path, out)
return False
except FileNotFoundError:
return True
def isfile(self, path): # pylint: disable=arguments-renamed
try:
meta = self.metadata(path)
return meta.isfile
return self.info(path)["type"] == "file"
except FileNotFoundError:
return False
@@ -192,19 +173,19 @@ class DvcFileSystem(FileSystem): # pylint:disable=abstract-method
assert topdown
root = os.path.abspath(top)
try:
meta = self.metadata(root)
info = self.info(root)
except FileNotFoundError:
if onerror is not None:
onerror(FileNotFoundError(top))
return
if not meta.isdir:
if info["type"] != "directory":
if onerror is not None:
onerror(NotADirectoryError(top))
return
trie = Trie()
for out in meta.outs:
for out in info["outs"]:
trie[out.fs.path.parts(out.fs_path)] = out
if out.is_dir_checksum and self.path.isin_or_eq(root, out.fs_path):
@@ -220,40 +201,64 @@ class DvcFileSystem(FileSystem): # pylint:disable=abstract-method
def isdvc(self, path, recursive=False, strict=True):
try:
meta = self.metadata(path)
info = self.info(path)
except FileNotFoundError:
return False
recurse = recursive or not strict
return meta.output_exists if recurse else meta.is_output
return bool(info.get("outs") if recurse else info.get("isout"))
def metadata(self, fs_path):
abspath = os.path.abspath(fs_path)
def info(self, path):
abspath = os.path.abspath(path)
try:
outs = self._find_outs(abspath, strict=False, recursive=True)
except OutputNotFoundError as exc:
raise FileNotFoundError from exc
meta = Metadata(fs_path=abspath, outs=outs, repo=self.repo)
meta.isdir = meta.isdir or self.check_isdir(meta.fs_path, meta.outs)
return meta
ret = {
"type": "file",
"outs": outs,
"size": 0,
"isexec": False,
"isdvc": False,
}
def info(self, path):
meta = self.metadata(path)
ret = {"type": "directory" if meta.isdir else "file"}
if meta.is_output and len(meta.outs) == 1 and meta.outs[0].hash_info:
out = meta.outs[0]
if len(outs) > 1:
ret["type"] = "directory"
return ret
out = outs[0]
if not out.hash_info:
ret["isexec"] = out.meta.isexec
return ret
if abspath == out.fs_path:
if out.hash_info.isdir:
ret["type"] = "directory"
ret["size"] = out.meta.size
ret["isexec"] = out.meta.isexec
ret[out.hash_info.name] = out.hash_info.value
elif meta.part_of_output:
(out,) = meta.outs
key = self.path.parts(self.path.relpath(path, out.fs_path))
(obj_meta, oid) = out.obj.trie.get(key) or (None, None)
if oid:
ret["size"] = obj_meta.size if obj_meta else 0
ret[oid.name] = oid.value
ret["isdvc"] = True
ret["isout"] = True
return ret
if out.fs_path.startswith(abspath + self.sep):
ret["type"] = "directory"
return ret
ret["isdvc"] = True
try:
self._get_granular_hash(abspath, out)
except FileNotFoundError:
ret["type"] = "directory"
return ret
key = self.repo.fs.path.relparts(abspath, out.fs_path)
(_, oid) = out.obj.trie.get(key) or (None, None)
ret[oid.name] = oid.value
return ret
def get_file(
+36 -48
View File
@@ -1,8 +1,6 @@
import errno
import logging
import os
import threading
from contextlib import suppress
from itertools import takewhile
from typing import TYPE_CHECKING, Callable, Optional, Tuple, Type, Union
@@ -66,7 +64,7 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
self._main_repo = repo
self.hash_jobs = repo.fs.hash_jobs
self.root_dir = repo.root_dir
self.root_dir: str = repo.root_dir
self._traverse_subrepos = subrepos
self._subrepos_trie = PathStringTrie()
@@ -234,11 +232,11 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
return True
try:
meta = dvc_fs.metadata(path)
info = dvc_fs.info(path)
except FileNotFoundError:
return False
for out in meta.outs:
for out in info["outs"]:
if fs.exists(out.fs_path):
return False
@@ -261,15 +259,15 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
return False
try:
meta = dvc_fs.metadata(path)
info = dvc_fs.info(path)
except FileNotFoundError:
return False
for out in meta.outs:
for out in info["outs"]:
if fs.exists(out.fs_path):
return False
return meta.isdir
return info["type"] == "directory"
def isdvc(self, path, **kwargs):
_, dvc_fs = self._get_fs_pair(path)
@@ -292,15 +290,16 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
return False
try:
meta = dvc_fs.metadata(path)
info = dvc_fs.info(path)
except FileNotFoundError:
return False
(out,) = meta.outs
assert len(meta.outs) == 1
(out,) = info["outs"]
assert len(info["outs"]) == 1
if fs.exists(out.fs_path):
return False
return meta.isfile
return info["type"] == "file"
def _dvc_walk(self, walk):
try:
@@ -455,47 +454,36 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
from_info, to_file, callback=callback, **kwargs
)
def metadata(self, path):
fs_path = os.path.abspath(path)
fs, dvc_fs = self._get_fs_pair(fs_path)
dvc_meta = None
if dvc_fs:
with suppress(FileNotFoundError):
dvc_meta = dvc_fs.metadata(fs_path)
info_result = None
with suppress(FileNotFoundError):
info_result = fs.info(fs_path)
if not info_result and not dvc_meta:
raise FileNotFoundError(
errno.ENOENT, os.strerror(errno.ENOENT), fs_path
)
from ._metadata import Metadata
meta = dvc_meta or Metadata(
fs_path=fs_path,
repo=self._get_repo(fs_path) or self._main_repo,
)
isdir = bool(info_result) and info_result["type"] == "directory"
meta.isdir = meta.isdir or isdir
if not dvc_meta:
from dvc.utils import is_exec
meta.is_exec = bool(info_result) and is_exec(info_result["mode"])
return meta
def info(self, path):
fs, dvc_fs = self._get_fs_pair(path)
try:
return fs.info(path)
dvc_info = dvc_fs.info(path)
except FileNotFoundError:
return dvc_fs.info(path)
dvc_info = None
try:
from dvc.utils import is_exec
fs_info = fs.info(path)
fs_info["repo"] = dvc_fs.repo
fs_info["outs"] = dvc_info.get("outs", None) if dvc_info else None
fs_info["isout"] = (
dvc_info.get("isout", False) if dvc_info else False
)
fs_info["isdvc"] = dvc_info["isdvc"] if dvc_info else False
fs_info["isexec"] = (
dvc_info["isexec"] if dvc_info else is_exec(fs_info["mode"])
)
return fs_info
except FileNotFoundError:
if not dvc_info:
raise
dvc_info["repo"] = dvc_fs.repo
dvc_info["isdvc"] = True
return dvc_info
def checksum(self, path):
fs, dvc_fs = self._get_fs_pair(path)
+4 -4
View File
@@ -189,9 +189,9 @@ def _dir_output_paths(fs, fs_path, obj, targets=None):
def _filter_missing(repo_fs, paths):
for path in paths:
try:
metadata = repo_fs.metadata(path)
if metadata.is_dvc:
out = metadata.outs[0]
info = repo_fs.info(path)
if info["isdvc"]:
out = info["outs"][0]
if out.status().get(str(out)) == "not in cache":
yield path
except FileNotFoundError:
@@ -204,7 +204,7 @@ def _targets_to_paths(repo_fs, targets):
for target in targets:
if repo_fs.exists(target):
paths.append(repo_fs.metadata(target).fs_path)
paths.append(os.path.abspath(target))
else:
missing.append(target)
+6 -8
View File
@@ -47,8 +47,6 @@ def ls(url, path=None, rev=None, recursive=None, dvc_only=False):
def _ls(repo, fs_path, recursive=None, dvc_only=False):
from dvc.fs._metadata import Metadata
def onerror(exc):
raise exc
@@ -70,20 +68,20 @@ def _ls(repo, fs_path, recursive=None, dvc_only=False):
ret = {}
for info in infos:
try:
metadata = fs.metadata(info)
_info = fs.info(info)
except FileNotFoundError:
# broken symlink
metadata = Metadata(info, repo)
_info = {"type": "file", "isexec": False}
if metadata.output_exists or not dvc_only:
if _info.get("outs") or not dvc_only:
path = (
fs.path.name(fs_path)
if fs_path == info
else fs.path.relpath(info, fs_path)
)
ret[path] = {
"isout": metadata.is_output,
"isdir": metadata.isdir,
"isexec": metadata.is_exec,
"isout": _info.get("isout", False),
"isdir": _info["type"] == "directory",
"isexec": _info["isexec"],
}
return ret
+175
View File
@@ -0,0 +1,175 @@
import os
import pytest
from dvc.fs.repo import RepoFileSystem
from tests.unit.fs.test_repo import make_subrepo
@pytest.fixture
def repo_fs(tmp_dir, dvc, scm):
fs_structure = {
"models": { # mixed dvc + git directory
"train.py": "train dot py",
"test.py": "test dot py",
},
"README.md": "my little project", # file
"src": { # repo-only directory
"utils": {
"__init__.py": "",
"serve_model.py": "# this will serve a model `soon`",
}
},
}
dvc_structure = {
"data": { # dvc only directory
"raw": {
"raw-1.csv": "one, dot, csv",
"raw-2.csv": "two, dot, csv",
},
"processed": {
"processed-1.csv": "1, dot, csv",
"processed-2.csv": "2, dot, csv",
},
},
os.path.join("models", "transform.pickle"): "model model", # file
}
tmp_dir.scm_gen(fs_structure, commit="repo init")
tmp_dir.dvc_gen(dvc_structure, commit="use dvc")
yield RepoFileSystem(dvc, subrepos=True)
def test_info_not_existing(repo_fs):
path = os.path.join("path", "that", "does", "not", "exist")
with pytest.raises(FileNotFoundError):
repo_fs.info(path)
@pytest.mark.parametrize(
"path",
[
"README.md",
os.path.join("models", "train.py"),
os.path.join("models", "test.py"),
os.path.join("src", "utils", "__init__.py"),
os.path.join("src", "utils", "serve_model.py"),
],
)
def test_info_git_tracked_file(repo_fs, path):
info = repo_fs.info(path)
assert info["repo"].root_dir == repo_fs.root_dir
assert not info["isdvc"]
assert info["type"] == "file"
assert not info["isexec"]
assert not info["outs"]
@pytest.mark.parametrize(
"path, outs",
[
(os.path.join("data", "raw", "raw-1.csv"), ["data"]),
(os.path.join("data", "raw", "raw-2.csv"), ["data"]),
(
os.path.join("data", "processed", "processed-1.csv"),
["data"],
),
(
os.path.join("data", "processed", "processed-2.csv"),
["data"],
),
(
os.path.join("models", "transform.pickle"),
[os.path.join("models", "transform.pickle")],
),
],
)
def test_info_dvc_tracked_file(repo_fs, path, outs):
info = repo_fs.info(path)
assert info["repo"].root_dir == repo_fs.root_dir
assert info["isdvc"]
assert info["type"] == "file"
assert not info["isexec"]
assert {out.fs_path for out in info["outs"]} == {
os.path.join(repo_fs.root_dir, out) for out in outs
}
@pytest.mark.parametrize("path", ["src", os.path.join("src", "utils")])
def test_info_git_only_dirs(repo_fs, path):
info = repo_fs.info(path)
assert info["repo"].root_dir == repo_fs.root_dir
assert not info["isdvc"]
assert info["type"] == "directory"
assert info["isexec"]
assert not info["outs"]
@pytest.mark.parametrize(
"path, expected_outs",
[
(".", ["data", os.path.join("models", "transform.pickle")]),
("models", [os.path.join("models", "transform.pickle")]),
],
)
def test_info_git_dvc_mixed_dirs(repo_fs, path, expected_outs):
info = repo_fs.info(os.path.join(repo_fs.root_dir, path))
assert info["repo"].root_dir == repo_fs.root_dir
assert not info["isdvc"]
assert info["type"] == "directory"
assert not info["isexec"]
assert {out.fs_path for out in info["outs"]} == {
os.path.join(repo_fs.root_dir, out) for out in expected_outs
}
@pytest.mark.parametrize(
"path",
[
"data",
os.path.join("data", "raw"),
os.path.join("data", "processed"),
],
)
def test_info_dvc_only_dirs(repo_fs, path):
info = repo_fs.info(os.path.join(repo_fs.root_dir, path))
assert info["repo"].root_dir == repo_fs.root_dir
assert info["isdvc"]
assert info["type"] == "directory"
assert not info["isexec"]
assert {out.fs_path for out in info["outs"]} == {
os.path.join(repo_fs.root_dir, "data")
}
def test_info_on_subrepos(make_tmp_dir, tmp_dir, dvc, scm, repo_fs):
subrepo = tmp_dir / "subrepo"
make_subrepo(subrepo, scm)
with subrepo.chdir():
subrepo.scm_gen("foo", "foo", commit="add foo on subrepo")
subrepo.dvc_gen("foobar", "foobar", commit="add foobar on subrepo")
for path in [
"subrepo",
os.path.join("subrepo", "foo"),
os.path.join("subrepo", "foobar"),
]:
info = repo_fs.info(tmp_dir / path)
assert info["repo"].root_dir == str(
subrepo
), f"repo root didn't match for {path}"
# supports external outputs on top-level DVC repo
external_dir = make_tmp_dir("external-output")
external_dir.gen("bar", "bar")
dvc.add(str(external_dir / "bar"), external=True)
info = repo_fs.info((external_dir / "bar").fs_path)
assert info["repo"].root_dir == str(tmp_dir)
-210
View File
@@ -1,210 +0,0 @@
import os
import pytest
from dvc.fs.repo import RepoFileSystem
from tests.unit.fs.test_repo import make_subrepo
@pytest.fixture
def repo_fs(tmp_dir, dvc, scm):
fs_structure = {
"models": { # mixed dvc + git directory
"train.py": "train dot py",
"test.py": "test dot py",
},
"README.md": "my little project", # file
"src": { # repo-only directory
"utils": {
"__init__.py": "",
"serve_model.py": "# this will serve a model `soon`",
}
},
}
dvc_structure = {
"data": { # dvc only directory
"raw": {
"raw-1.csv": "one, dot, csv",
"raw-2.csv": "two, dot, csv",
},
"processed": {
"processed-1.csv": "1, dot, csv",
"processed-2.csv": "2, dot, csv",
},
},
os.path.join("models", "transform.pickle"): "model model", # file
}
tmp_dir.scm_gen(fs_structure, commit="repo init")
tmp_dir.dvc_gen(dvc_structure, commit="use dvc")
yield RepoFileSystem(dvc, subrepos=True)
def test_metadata_not_existing(repo_fs):
path = os.path.join("path", "that", "does", "not", "exist")
with pytest.raises(FileNotFoundError):
repo_fs.metadata(path)
@pytest.mark.parametrize(
"path",
[
"README.md",
os.path.join("models", "train.py"),
os.path.join("models", "test.py"),
os.path.join("src", "utils", "__init__.py"),
os.path.join("src", "utils", "serve_model.py"),
],
)
def test_metadata_git_tracked_file(repo_fs, path):
meta = repo_fs.metadata(path)
assert meta.fs_path == os.path.join(repo_fs.root_dir, path)
assert meta.repo.root_dir == repo_fs.root_dir
assert not meta.is_output
assert not meta.part_of_output
assert not meta.contains_outputs
assert not meta.is_dvc
assert not meta.output_exists
assert not meta.isdir
assert not meta.is_exec
assert meta.isfile
assert not meta.outs
@pytest.mark.parametrize(
"path, outs, is_output",
[
(os.path.join("data", "raw", "raw-1.csv"), ["data"], False),
(os.path.join("data", "raw", "raw-2.csv"), ["data"], False),
(
os.path.join("data", "processed", "processed-1.csv"),
["data"],
False,
),
(
os.path.join("data", "processed", "processed-2.csv"),
["data"],
False,
),
(
os.path.join("models", "transform.pickle"),
[os.path.join("models", "transform.pickle")],
True,
),
],
)
def test_metadata_dvc_tracked_file(repo_fs, path, outs, is_output):
meta = repo_fs.metadata(path)
assert meta.fs_path == os.path.join(repo_fs.root_dir, path)
assert meta.repo.root_dir == repo_fs.root_dir
assert meta.is_output == is_output
assert meta.part_of_output != is_output
assert not meta.contains_outputs
assert meta.is_dvc
assert meta.output_exists
assert not meta.isdir
assert not meta.is_exec
assert meta.isfile
assert {out.fs_path for out in meta.outs} == {
os.path.join(repo_fs.root_dir, out) for out in outs
}
@pytest.mark.parametrize("path", ["src", os.path.join("src", "utils")])
def test_metadata_git_only_dirs(repo_fs, path):
meta = repo_fs.metadata(path)
assert meta.fs_path == os.path.join(repo_fs.root_dir, path)
assert meta.repo.root_dir == repo_fs.root_dir
assert not meta.is_output
assert not meta.part_of_output
assert not meta.contains_outputs
assert not meta.is_dvc
assert not meta.output_exists
assert meta.isdir
assert meta.is_exec
assert not meta.isfile
assert not meta.outs
@pytest.mark.parametrize(
"path, expected_outs",
[
(".", ["data", os.path.join("models", "transform.pickle")]),
("models", [os.path.join("models", "transform.pickle")]),
],
)
def test_metadata_git_dvc_mixed_dirs(repo_fs, path, expected_outs):
meta = repo_fs.metadata(os.path.join(repo_fs.root_dir, path))
assert meta.fs_path == os.path.normpath(
os.path.join(repo_fs.root_dir, path)
)
assert meta.repo.root_dir == repo_fs.root_dir
assert not meta.is_output
assert not meta.part_of_output
assert meta.contains_outputs
assert not meta.is_dvc
assert meta.output_exists
assert meta.isdir
assert not meta.is_exec
assert not meta.isfile
assert {out.fs_path for out in meta.outs} == {
os.path.join(repo_fs.root_dir, out) for out in expected_outs
}
@pytest.mark.parametrize(
"path, is_output",
[
("data", True),
(os.path.join("data", "raw"), False), # is inside output
(os.path.join("data", "processed"), False),
],
)
def test_metadata_dvc_only_dirs(repo_fs, path, is_output):
meta = repo_fs.metadata(os.path.join(repo_fs.root_dir, path))
assert meta.fs_path == os.path.join(repo_fs.root_dir, path)
assert meta.repo.root_dir == repo_fs.root_dir
assert meta.is_output == is_output
assert meta.part_of_output != is_output
assert not meta.contains_outputs
assert meta.is_dvc
assert meta.output_exists
assert meta.isdir
assert not meta.is_exec
assert not meta.isfile
assert {out.fs_path for out in meta.outs} == {
os.path.join(repo_fs.root_dir, "data")
}
def test_metadata_on_subrepos(make_tmp_dir, tmp_dir, dvc, scm, repo_fs):
subrepo = tmp_dir / "subrepo"
make_subrepo(subrepo, scm)
with subrepo.chdir():
subrepo.scm_gen("foo", "foo", commit="add foo on subrepo")
subrepo.dvc_gen("foobar", "foobar", commit="add foobar on subrepo")
for path in [
"subrepo",
os.path.join("subrepo", "foo"),
os.path.join("subrepo", "foobar"),
]:
meta = repo_fs.metadata(tmp_dir / path)
assert meta.repo.root_dir == str(
subrepo
), f"repo root didn't match for {path}"
# supports external outputs on top-level DVC repo
external_dir = make_tmp_dir("external-output")
external_dir.gen("bar", "bar")
dvc.add(str(external_dir / "bar"), external=True)
meta = repo_fs.metadata((external_dir / "bar").fs_path)
assert meta.repo.root_dir == str(tmp_dir)