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.

Author: efiop

dvc/api.py

@@ -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)
 
 

dvc/fs/_metadata.py

@@ -6,7 +6,6 @@
 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 @@ def open(  # type: ignore
 
     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 @@ def walk(self, top, topdown=True, onerror=None, **kwargs):
         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 @@ def find(self, path, prefix=None):
 
     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(

dvc/fs/dvc.py

@@ -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 @@ def __init__(
 
         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 @@ def exists(self, path) -> bool:
             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 @@ def isdir(self, path):  # pylint: disable=arguments-renamed
             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 @@ def isfile(self, path):  # pylint: disable=arguments-renamed
             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 @@ def get_file(
             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,
-        )
+    def info(self, path):
+        fs, dvc_fs = self._get_fs_pair(path)
 
-        isdir = bool(info_result) and info_result["type"] == "directory"
-        meta.isdir = meta.isdir or isdir
+        try:
+            dvc_info = dvc_fs.info(path)
+        except FileNotFoundError:
+            dvc_info = None
 
-        if not dvc_meta:
+        try:
             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)
+            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
 
-        try:
-            return fs.info(path)
         except FileNotFoundError:
-            return dvc_fs.info(path)
+            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)