dvc: bump to 2.5.4

Kudos @skshetry for discovering that we've been publishing py2.py3
wheels on pypi and researching the cause.

Co-authored-by: Saugat Pachhai <[email protected]>

Co-authored-by: Saugat Pachhai <[email protected]>

.pre-commit-config.yaml

@@ -7,6 +7,13 @@ repos:
         language_version: python3
     repo: https://github.com/ambv/black
     rev: 21.6b0
+  - hooks:
+      - id: codespell
+        args: 
+          - --ignore-words-list
+          - ba,datas,fo,uptodate
+    repo: https://github.com/codespell-project/codespell
+    rev: v2.1.0
   - hooks:
       - id: isort
         language_version: python3

README.rst

@@ -184,7 +184,7 @@ Comparison to related technologies
 
 #. *Data Engineering* tools such as `AirFlow <https://airflow.apache.org/>`_, `Luigi <https://github.com/spotify/luigi>`_ and
    others - in DVC data, model and ML pipelines represent a single ML project and focuses on data scientist' experience while
-   data engineering pipeline orchestrates multiple data projects and focuses on efficent execution. DVC project can be used
+   data engineering pipeline orchestrates multiple data projects and focuses on efficient execution. DVC project can be used
    from data pipeline as a single execution step. `AirFlow DVC <https://github.com/covid-genomics/airflow-dvc>`_ is an example
    of such integration.
 

dvc/command/diff.py

@@ -111,14 +111,12 @@ class CmdDiff(CmdBase):
         if not sum(summary.values()):
             return None
 
-        fmt = (
-            "files summary: {added} added, {deleted} deleted,"
-            " {renamed} renamed, {modified} modified"
+        states_summary = ", ".join(
+            f"{summary[state]} {state}"
+            for state in states
+            if summary[state] > 0
         )
-        if not hide_missing:
-            fmt += ", {not in cache} not in cache"
-
-        ui.write(fmt.format_map(summary))
+        ui.write("files summary:", states_summary)
 
     def run(self):
         from dvc.exceptions import DvcException

dvc/dagascii.py

@@ -163,7 +163,7 @@ class AsciiCanvas:
         self.point(x0 + width, y0 + height, "+")
 
 
-def _build_sugiyama_layout(vertexes, edges):
+def _build_sugiyama_layout(vertices, edges):
     #
     # Just a reminder about naming conventions:
     # +------------X
@@ -174,17 +174,17 @@ def _build_sugiyama_layout(vertexes, edges):
     # Y
     #
 
-    vertexes = {v: Vertex(f" {v} ") for v in vertexes}
+    vertices = {v: Vertex(f" {v} ") for v in vertices}
     # NOTE: reverting edges to correctly orientate the graph
-    edges = [Edge(vertexes[e], vertexes[s]) for s, e in edges]
-    vertexes = vertexes.values()
-    graph = Graph(vertexes, edges)
+    edges = [Edge(vertices[e], vertices[s]) for s, e in edges]
+    vertices = vertices.values()
+    graph = Graph(vertices, edges)
 
-    for vertex in vertexes:
+    for vertex in vertices:
         vertex.view = VertexViewer(vertex.data)
 
     # NOTE: determine min box length to create the best layout
-    minw = min(v.view.w for v in vertexes)
+    minw = min(v.view.w for v in vertices)
 
     for edge in edges:
         edge.view = EdgeViewer()
@@ -204,11 +204,11 @@ def _build_sugiyama_layout(vertexes, edges):
     return sug
 
 
-def draw(vertexes, edges):
+def draw(vertices, edges):
     """Build a DAG and draw it in ASCII.
 
     Args:
-        vertexes (list): list of graph vertexes.
+        vertices (list): list of graph vertices.
         edges (list): list of graph edges.
     """
     # pylint: disable=too-many-locals
@@ -217,7 +217,7 @@ def draw(vertexes, edges):
     Xs = []  # pylint: disable=invalid-name
     Ys = []  # pylint: disable=invalid-name
 
-    sug = _build_sugiyama_layout(vertexes, edges)
+    sug = _build_sugiyama_layout(vertices, edges)
 
     for vertex in sug.g.sV:
         # NOTE: moving boxes w/2 to the left

dvc/fs/azure.py

@@ -25,7 +25,7 @@ _DEFAULT_CREDS_STEPS = (
 
 @contextmanager
 def _temp_event_loop():
-    """When trying to initalize azure filesystem instances
+    """When trying to initialize azure filesystem instances
     with DefaultCredentials, the authentication process requires
     to have an access to a separate event loop. The normal calls
     are run in a separate loop managed by the fsspec, but the

dvc/fs/dvc.py

@@ -233,7 +233,7 @@ class DvcFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
 
     def info(self, path_info):
         meta = self.metadata(path_info)
-        ret = {"type": "dir" if meta.isdir else "file"}
+        ret = {"type": "directory" if meta.isdir else "file"}
         if meta.is_output and len(meta.outs) == 1 and meta.outs[0].hash_info:
             hash_info = meta.outs[0].hash_info
             ret["size"] = hash_info.size

dvc/fs/fsspec_wrapper.py

@@ -48,7 +48,7 @@ class FSSpecWrapper(BaseFileSystem):
             yield entry
 
     def _entry_hook(self, entry):
-        """Simple hook method to be overriden when wanted to process
+        """Simple hook method to be overridden when wanted to process
         entries within info() and ls(detail=True) calls"""
         return entry
 

dvc/fs/gdrive.py

@@ -1,3 +1,4 @@
+import errno
 import io
 import logging
 import os
@@ -11,7 +12,7 @@ from urllib.parse import urlparse
 from funcy import cached_property, retry, wrap_prop, wrap_with
 from funcy.py3 import cat
 
-from dvc.exceptions import DvcException, FileMissingError
+from dvc.exceptions import DvcException
 from dvc.path_info import CloudURLInfo
 from dvc.progress import Tqdm
 from dvc.scheme import Schemes
@@ -521,12 +522,14 @@ class GDriveFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
             return min(item_ids)
 
         assert not create
-        raise FileMissingError(path_info, hint)
+        raise FileNotFoundError(
+            errno.ENOENT, os.strerror(errno.ENOENT), hint or path_info
+        )
 
     def exists(self, path_info) -> bool:
         try:
             self._get_item_id(path_info)
-        except FileMissingError:
+        except FileNotFoundError:
             return False
         else:
             return True
@@ -616,7 +619,7 @@ class GDriveFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
         item_id = self._get_item_id(path_info)
         gdrive_file = self._drive.CreateFile({"id": item_id})
         gdrive_file.FetchMetadata(fields="fileSize")
-        return {"size": gdrive_file.get("fileSize")}
+        return {"size": gdrive_file.get("fileSize"), "type": "file"}
 
     def _upload_fobj(self, fobj, to_info, **kwargs):
         dirname = to_info.parent

dvc/fs/git.py

@@ -1,5 +1,6 @@
 import errno
 import os
+import stat
 
 from dvc.utils import is_exec, relpath
 
@@ -106,7 +107,10 @@ class GitFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
             raise FileNotFoundError(
                 errno.ENOENT, os.strerror(errno.ENOENT), path_info
             )
-        return {"size": st.st_size}
+        return {
+            "size": st.st_size,
+            "type": "directory" if stat.S_ISDIR(st.st_mode) else "file",
+        }
 
     def stat(self, path):
         key = self._get_key(path)

dvc/fs/hdfs.py

@@ -1,3 +1,4 @@
+import errno
 import io
 import logging
 import os
@@ -220,8 +221,19 @@ class HDFSFileSystem(BaseFileSystem):
 
     def info(self, path_info):
         with self.hdfs(path_info) as hdfs:
-            finfo = hdfs.get_file_info(path_info.path)
-            return {"size": finfo.size}
+            from pyarrow.fs import FileType
+
+            file_info = hdfs.get_file_info(path_info.path)
+            if file_info.type is FileType.Directory:
+                kind = "directory"
+            elif file_info.type is FileType.File:
+                kind = "file"
+            else:
+                raise FileNotFoundError(
+                    errno.ENOENT, os.strerror(errno.ENOENT), path_info
+                )
+
+            return {"size": file_info.size, "type": kind}
 
     def checksum(self, path_info):
         return HashInfo(

dvc/fs/http.py

@@ -31,6 +31,7 @@ class HTTPFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
     PATH_CLS = HTTPURLInfo
     PARAM_CHECKSUM = "etag"
     CAN_TRAVERSE = False
+    REQUIRES = {"requests": "requests"}
 
     SESSION_RETRIES = 5
     SESSION_BACKOFF_FACTOR = 0.1
@@ -145,7 +146,7 @@ class HTTPFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
         resp = self._head(path_info.url)
         etag = resp.headers.get("ETag") or resp.headers.get("Content-MD5")
         size = self._content_length(resp)
-        return {"etag": etag, "size": size}
+        return {"etag": etag, "size": size, "type": "file"}
 
     def _upload_fobj(self, fobj, to_info, **kwargs):
         def chunks(fobj):

dvc/fs/local.py

@@ -157,7 +157,7 @@ class LocalFileSystem(BaseFileSystem):
         st = os.stat(path_info)
         return {
             "size": st.st_size,
-            "type": "dir" if stat.S_ISDIR(st.st_mode) else "file",
+            "type": "directory" if stat.S_ISDIR(st.st_mode) else "file",
         }
 
     def _upload(

dvc/fs/s3.py

@@ -188,6 +188,7 @@ class S3FileSystem(BaseS3FileSystem):  # pylint:disable=abstract-method
             "s3",
             endpoint_url=client_kwargs.get("endpoint_url"),
             use_ssl=login_info["use_ssl"],
+            verify=client_kwargs.get("verify"),
         )
 
     def _get_obj(self, path_info):

dvc/fs/ssh/connection.py

@@ -82,7 +82,7 @@ class SSHConnection:
         st = self.sftp.stat(path)
         return {
             "size": st.st_size,
-            "type": "dir" if stat.S_ISDIR(st.st_mode) else "file",
+            "type": "directory" if stat.S_ISDIR(st.st_mode) else "file",
         }
 
     def getsize(self, path):

dvc/fs/webhdfs.py

@@ -123,7 +123,7 @@ class WebHDFSFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
 
     def info(self, path_info):
         st = self.hdfs_client.status(path_info.path)
-        return {"size": st["length"]}
+        return {"size": st["length"], "type": "file"}
 
     def checksum(self, path_info):
         return HashInfo(

dvc/info.py

@@ -20,6 +20,11 @@ if PKG is None:
 else:
     package = f"({PKG})"
 
+try:
+    import importlib.metadata as importlib_metadata
+except ImportError:  # < 3.8
+    import importlib_metadata  # type: ignore[no-redef]
+
 
 def get_dvc_info():
     info = [
@@ -27,7 +32,7 @@ def get_dvc_info():
         "---------------------------------",
         f"Platform: Python {platform.python_version()} on "
         f"{platform.platform()}",
-        f"Supports: {_get_supported_remotes()}",
+        f"Supports:{_get_supported_remotes()}",
     ]
 
     try:
@@ -113,19 +118,23 @@ def _get_linktype_support_info(repo):
 
 
 def _get_supported_remotes():
-
     supported_remotes = []
     for scheme, fs_cls in FS_MAP.items():
         if not fs_cls.get_missing_deps():
-            supported_remotes.append(scheme)
+            dependencies = []
+            for requirement in fs_cls.REQUIRES:
+                dependencies.append(
+                    f"{requirement} = "
+                    f"{importlib_metadata.version(requirement)}"
+                )
 
-    if len(supported_remotes) == len(FS_MAP):
-        return "All remotes"
+            remote_info = scheme
+            if dependencies:
+                remote_info += " (" + ", ".join(dependencies) + ")"
+            supported_remotes.append(remote_info)
 
-    if len(supported_remotes) == 1:
-        return supported_remotes
-
-    return ", ".join(supported_remotes)
+    assert len(supported_remotes) >= 1
+    return "\n\t" + ",\n\t".join(supported_remotes)
 
 
 def get_fs_type(path):

dvc/objects/file.py

@@ -28,9 +28,7 @@ class HashFile:
 
     @property
     def size(self):
-        if not (self.path_info and self.fs):
-            return None
-        return self.fs.getsize(self.path_info)
+        return self.hash_info.size
 
     def __len__(self):
         return 1

dvc/objects/stage.py

@@ -1,5 +1,3 @@
-import errno
-import os
 from concurrent.futures import ThreadPoolExecutor
 from functools import partial
 
@@ -174,11 +172,7 @@ def stage(odb, path_info, fs, name, upload=False, **kwargs):
         isinstance(path_info, str) or path_info.scheme == fs.scheme
     )
 
-    if not fs.exists(path_info):
-        raise FileNotFoundError(
-            errno.ENOENT, os.strerror(errno.ENOENT), path_info
-        )
-
+    details = fs.info(path_info)
     state = odb.state
     # pylint: disable=assignment-from-none
     hash_info = state.get(path_info, fs)
@@ -210,12 +204,12 @@ def stage(odb, path_info, fs, name, upload=False, **kwargs):
         obj.hash_info.size = hash_info.size
         return obj
 
-    if fs.isdir(path_info):
+    if details["type"] == "directory":
         obj = _get_tree_obj(path_info, fs, name, odb, state, upload, **kwargs)
     else:
         _, obj = _get_file_obj(path_info, fs, name, odb, state, upload)
 
-    if obj.hash_info and fs.exists(path_info):
+    if obj.hash_info:
         state.save(path_info, fs, obj.hash_info)
 
     return obj

dvc/objects/tree.py

@@ -28,13 +28,6 @@ class Tree(HashFile):
 
         return Trie(self._dict)
 
-    @property
-    def size(self):
-        try:
-            return sum(obj.size for _, obj in self)
-        except TypeError:
-            return None
-
     def add(self, key, obj):
         self.__dict__.pop("trie", None)
         self._dict[key] = obj
@@ -52,7 +45,10 @@ class Tree(HashFile):
         self.path_info = path_info
         self.hash_info = get_file_hash(path_info, memfs, "md5")
         self.hash_info.value += ".dir"
-        self.hash_info.size = self.size
+        try:
+            self.hash_info.size = sum(obj.size for _, obj in self)
+        except TypeError:
+            self.hash_info.size = None
         self.hash_info.nfiles = len(self)
 
     def __len__(self):

dvc/progress.py

@@ -51,7 +51,7 @@ class Tqdm(tqdm):
         file=None,
         total=None,
         postfix=None,
-        **kwargs
+        **kwargs,
     ):
         """
         bytes   : shortcut for
@@ -90,7 +90,7 @@ class Tqdm(tqdm):
             bar_format="!",
             lock_args=(False,),
             total=total,
-            **kwargs
+            **kwargs,
         )
         self.postfix = postfix or {"info": ""}
         if bar_format is None:
@@ -106,13 +106,16 @@ class Tqdm(tqdm):
             self.bar_format = bar_format
         self.refresh()
 
-    def update_msg(self, msg, n=1):
+    def update_msg(self, msg: str, n: int = 1) -> None:
         """
         Sets `msg` as a postfix and calls `update(n)`.
         """
-        self.postfix["info"] = " %s |" % msg
+        self.set_msg(msg)
         self.update(n)
 
+    def set_msg(self, msg: str) -> None:
+        self.postfix["info"] = f" {msg} |"
+
     def update_to(self, current, total=None):
         if total:
             self.total = total

dvc/repo/add.py

@@ -1,9 +1,13 @@
 import logging
 import os
-from typing import TYPE_CHECKING
+from contextlib import contextmanager
+from itertools import tee
+from typing import TYPE_CHECKING, Any, Iterator, List
 
 import colorama
 
+from dvc.ui import ui
+
 from ..exceptions import (
     CacheLinkError,
     InvalidArgumentError,
@@ -11,19 +15,28 @@ from ..exceptions import (
     OverlappingOutputPathsError,
     RecursiveAddingWhileUsingFilename,
 )
-from ..progress import Tqdm
 from ..repo.scm_context import scm_context
 from ..utils import LARGE_DIR_SIZE, glob_targets, resolve_output, resolve_paths
 from ..utils.collections import ensure_list, validate
 from . import locked
 
 if TYPE_CHECKING:
+    from dvc.repo import Repo
+    from dvc.stage import Stage
     from dvc.types import TargetType
 
-
+Stages = List["Stage"]
 logger = logging.getLogger(__name__)
 
 
+OVERLAPPING_OUTPUT_FMT = (
+    "Cannot add '{out}', because it is overlapping with other "
+    "DVC tracked output: '{parent}'.\n"
+    "To include '{out}' in '{parent}', run "
+    "'dvc commit {parent_stage}'"
+)
+
+
 def check_recursive_and_fname(args):
     if args.recursive and args.fname:
         raise RecursiveAddingWhileUsingFilename()
@@ -61,6 +74,63 @@ def check_arg_combinations(args):
         raise InvalidArgumentError(message.format(option=invalid_opt))
 
 
+@contextmanager
+def translate_graph_error(stages: Stages) -> Iterator[None]:
+    try:
+        yield
+    except OverlappingOutputPathsError as exc:
+        msg = OVERLAPPING_OUTPUT_FMT.format(
+            out=exc.overlapping_out.path_info,
+            parent=exc.parent.path_info,
+            parent_stage=exc.parent.stage.addressing,
+        )
+        raise OverlappingOutputPathsError(exc.parent, exc.overlapping_out, msg)
+    except OutputDuplicationError as exc:
+        raise OutputDuplicationError(
+            exc.output, list(set(exc.stages) - set(stages))
+        )
+
+
+def progress_iter(stages: Stages) -> Iterator["Stage"]:
+    total = len(stages)
+    desc = "Adding..."
+    with ui.progress(
+        stages, total=total, desc=desc, unit="file", leave=True
+    ) as pbar:
+        if total == 1:
+            pbar.bar_format = desc
+            pbar.refresh()
+
+        for stage in pbar:
+            if total > 1:
+                pbar.set_msg(f"{stage.outs[0]}")
+            yield stage
+            if total == 1:  # restore bar format for stats
+                # pylint: disable=no-member
+                pbar.bar_format = pbar.BAR_FMT_DEFAULT
+
+
+LINK_FAILURE_MESSAGE = (
+    "\nSome targets could not be linked from cache to workspace.\n{}\n"
+    "To re-link these targets, reconfigure cache types and then run:\n"
+    "\n\tdvc checkout {}"
+)
+
+
+@contextmanager
+def warn_link_failures() -> Iterator[List[str]]:
+    link_failures: List[str] = []
+    try:
+        yield link_failures
+    finally:
+        if link_failures:
+            msg = LINK_FAILURE_MESSAGE.format(
+                CacheLinkError.SUPPORT_LINK,
+                " ".join(link_failures),
+            )
+            ui.error_write(msg)
+
+
 VALIDATORS = (
     check_recursive_and_fname,
     transform_targets,
@@ -68,226 +138,127 @@ VALIDATORS = (
 )
 
 
+@validate(*VALIDATORS)
 @locked
 @scm_context
-@validate(*VALIDATORS)
 def add(  # noqa: C901
-    repo,
+    repo: "Repo",
     targets: "TargetType",
-    recursive=False,
-    no_commit=False,
-    fname=None,
-    to_remote=False,
-    **kwargs,
+    recursive: bool = False,
+    no_commit: bool = False,
+    fname: str = None,
+    to_remote: bool = False,
+    **kwargs: Any,
 ):
-    link_failures = []
-    stages_list = []
-    num_targets = len(targets)
-    to_cache = kwargs.get("out") and not to_remote
-    with Tqdm(total=num_targets, desc="Add", unit="file", leave=True) as pbar:
-        if num_targets == 1:
-            # clear unneeded top-level progress bar for single target
-            pbar.bar_format = "Adding..."
-            pbar.refresh()
-        for target in targets:
-            sub_targets = _find_all_targets(repo, target, recursive)
-            pbar.total += len(sub_targets) - 1
+    to_cache = bool(kwargs.get("out")) and not to_remote
+    transfer = to_remote or to_cache
 
-            if os.path.isdir(target) and len(sub_targets) > LARGE_DIR_SIZE:
-                logger.warning(
-                    "You are adding a large directory '{target}' recursively,"
-                    " consider tracking it as a whole instead.\n"
-                    "{purple}HINT:{nc} Remove the generated DVC file and then"
-                    " run `{cyan}dvc add {target}{nc}`".format(
-                        purple=colorama.Fore.MAGENTA,
-                        cyan=colorama.Fore.CYAN,
-                        nc=colorama.Style.RESET_ALL,
-                        target=target,
-                    )
+    glob = kwargs.get("glob", False)
+    add_targets = collect_targets(repo, targets, recursive, glob)
+    # pass one for creating stages, other one is used for iterating here
+    add_targets, sources = tee(add_targets)
+
+    # collect targets and build stages as we go
+    desc = "Collecting targets"
+    stages_it = create_stages(repo, add_targets, fname, transfer, **kwargs)
+    stages = list(ui.progress(stages_it, desc=desc, unit="file"))
+
+    msg = "Collecting stages from the workspace"
+    with translate_graph_error(stages), ui.status(msg) as status:
+        # remove existing stages that are to-be replaced with these
+        # new stages for the graph checks.
+        old_stages = set(repo.stages) - set(stages)
+        status.update("Checking graph")
+        repo.check_modified_graph(stages, list(old_stages))
+
+    odb = None
+    if to_remote:
+        remote = repo.cloud.get_remote(kwargs.get("remote"), "add")
+        odb = remote.odb
+
+    with warn_link_failures() as link_failures:
+        for stage, source in zip(progress_iter(stages), sources):
+            if to_remote or to_cache:
+                stage.transfer(source, to_remote=to_remote, odb=odb, **kwargs)
+            else:
+                try:
+                    stage.save()
+                    if not no_commit:
+                        stage.commit()
+                except CacheLinkError:
+                    link_failures.append(str(stage.relpath))
+            stage.dump()
+    return stages
+
+
+LARGE_DIR_RECURSIVE_ADD_WARNING = (
+    "You are adding a large directory '{target}' recursively.\n"
+    "Consider tracking it as a whole instead with "
+    "`{cyan}dvc add {target}{nc}`."
+)
+
+
+def collect_targets(
+    repo: "Repo",
+    targets: "TargetType",
+    recursive: bool = False,
+    glob: bool = False,
+) -> Iterator[str]:
+    for target in glob_targets(ensure_list(targets), glob=glob):
+        expanded_targets = _find_all_targets(repo, target, recursive=recursive)
+        for index, path in enumerate(expanded_targets):
+            if index == LARGE_DIR_SIZE:
+                msg = LARGE_DIR_RECURSIVE_ADD_WARNING.format(
+                    cyan=colorama.Fore.CYAN,
+                    nc=colorama.Style.RESET_ALL,
+                    target=target,
                 )
-
-            stages = _create_stages(
-                repo,
-                sub_targets,
-                fname,
-                pbar=pbar,
-                transfer=to_remote or to_cache,
-                **kwargs,
-            )
-
-            # remove existing stages that are to-be replaced with these
-            # new stages for the graph checks.
-            old_stages = set(repo.stages) - set(stages)
-            try:
-                repo.check_modified_graph(stages, list(old_stages))
-            except OverlappingOutputPathsError as exc:
-                msg = (
-                    "Cannot add '{out}', because it is overlapping with other "
-                    "DVC tracked output: '{parent}'.\n"
-                    "To include '{out}' in '{parent}', run "
-                    "'dvc commit {parent_stage}'"
-                ).format(
-                    out=exc.overlapping_out.path_info,
-                    parent=exc.parent.path_info,
-                    parent_stage=exc.parent.stage.addressing,
-                )
-                raise OverlappingOutputPathsError(
-                    exc.parent, exc.overlapping_out, msg
-                )
-            except OutputDuplicationError as exc:
-                raise OutputDuplicationError(
-                    exc.output, list(set(exc.stages) - set(stages))
-                )
-
-            link_failures.extend(
-                _process_stages(
-                    repo,
-                    sub_targets,
-                    stages,
-                    no_commit,
-                    pbar,
-                    to_remote,
-                    to_cache,
-                    **kwargs,
-                )
-            )
-            stages_list += stages
-
-        if num_targets == 1:  # restore bar format for stats
-            pbar.bar_format = pbar.BAR_FMT_DEFAULT
-
-    if link_failures:
-        msg = (
-            "Some targets could not be linked from cache to workspace.\n{}\n"
-            "To re-link these targets, reconfigure cache types and then run:\n"
-            "\n\tdvc checkout {}"
-        ).format(
-            CacheLinkError.SUPPORT_LINK,
-            " ".join([str(stage.relpath) for stage in link_failures]),
-        )
-        logger.warning(msg)
-
-    return stages_list
+                ui.error_write(msg)
+            yield path
 
 
-def _process_stages(
-    repo, sub_targets, stages, no_commit, pbar, to_remote, to_cache, **kwargs
-):
-    link_failures = []
-    from dvc.dvcfile import Dvcfile
-
-    from ..output import OutputDoesNotExistError
-
-    if to_remote or to_cache:
-        # Already verified in the add()
-        (stage,) = stages
-        (target,) = sub_targets
-        (out,) = stage.outs
-
-        odb = None
-        if to_remote:
-            remote = repo.cloud.get_remote(kwargs.get("remote"), "add")
-            odb = remote.odb
-
-        out.transfer(target, odb=odb, jobs=kwargs.get("jobs"))
-
-        if to_cache:
-            out.checkout()
-
-        Dvcfile(repo, stage.path).dump(stage)
-        return link_failures
-
-    with Tqdm(
-        total=len(stages),
-        desc="Processing",
-        unit="file",
-        disable=len(stages) == 1,
-    ) as pbar_stages:
-        for stage in stages:
-            try:
-                stage.save()
-            except OutputDoesNotExistError:
-                pbar.n -= 1
-                raise
-
-            try:
-                if not no_commit:
-                    stage.commit()
-            except CacheLinkError:
-                link_failures.append(stage)
-
-            Dvcfile(repo, stage.path).dump(stage)
-            pbar_stages.update()
-
-    return link_failures
-
-
-def _find_all_targets(repo, target, recursive):
+def _find_all_targets(
+    repo: "Repo", target: str, recursive: bool = False
+) -> Iterator[str]:
     from dvc.dvcfile import is_dvc_file
 
     if os.path.isdir(target) and recursive:
-        return [
-            os.fspath(path)
-            for path in Tqdm(
-                repo.dvcignore.walk_files(repo.fs, target),
-                desc="Searching " + target,
-                bar_format=Tqdm.BAR_FMT_NOTOTAL,
-                unit="file",
-            )
-            if not repo.is_dvc_internal(os.fspath(path))
-            if not is_dvc_file(os.fspath(path))
-            if not repo.scm.belongs_to_scm(os.fspath(path))
-            if not repo.scm.is_tracked(os.fspath(path))
-        ]
-    return [target]
-
-
-def _create_stages(
-    repo,
-    targets,
-    fname,
-    pbar=None,
-    external=False,
-    glob=False,
-    desc=None,
-    transfer=False,
-    **kwargs,
-):
-    from dvc.stage import Stage, create_stage, restore_meta
-
-    expanded_targets = glob_targets(targets, glob=glob)
-
-    stages = []
-    for out in Tqdm(
-        expanded_targets,
-        desc="Creating DVC files",
-        disable=len(expanded_targets) < LARGE_DIR_SIZE,
-        unit="file",
-    ):
-        if kwargs.get("out"):
-            out = resolve_output(out, kwargs["out"])
-        path, wdir, out = resolve_paths(
-            repo, out, always_local=transfer and not kwargs.get("out")
+        files = map(os.fspath, repo.dvcignore.walk_files(repo.fs, target))
+        yield from (
+            path
+            for path in files
+            if not repo.is_dvc_internal(path)
+            if not is_dvc_file(path)
+            if not repo.scm.belongs_to_scm(path)
+            if not repo.scm.is_tracked(path)
         )
-        stage = create_stage(
-            Stage,
-            repo,
-            fname or path,
+    else:
+        yield target
+
+
+def create_stages(
+    repo: "Repo",
+    targets: Iterator[str],
+    fname: str = None,
+    transfer: bool = False,
+    external: bool = False,
+    **kwargs: Any,
+) -> Iterator["Stage"]:
+    for target in targets:
+        if kwargs.get("out"):
+            target = resolve_output(target, kwargs["out"])
+        path, wdir, out = resolve_paths(
+            repo, target, always_local=transfer and not kwargs.get("out")
+        )
+
+        stage = repo.stage.create(
+            single_stage=True,
+            validate=False,
+            fname=fname or path,
             wdir=wdir,
             outs=[out],
             external=external,
         )
-        restore_meta(stage)
-        if desc:
-            stage.outs[0].desc = desc
-
-        if not stage:
-            if pbar is not None:
-                pbar.total -= 1
-            continue
-
-        stages.append(stage)
-        if pbar is not None:
-            pbar.update_msg(out)
-
-    return stages
+        if kwargs.get("desc"):
+            stage.outs[0].desc = kwargs["desc"]
+        yield stage

dvc/repo/experiments/__init__.py

@@ -753,7 +753,7 @@ class Experiments:
         assert entry.index == 0
 
         # NOTE: the stash commit to be popped already contains all the current
-        # workspace changes plus CLI modifed --params changes.
+        # workspace changes plus CLI modified --params changes.
         # `checkout --force` here will not lose any data (popping stash commit
         # will result in conflict between workspace params and stashed CLI
         # params, but we always want the stashed version).

dvc/scm/git/backend/base.py

@@ -187,7 +187,7 @@ class BaseGitBackend(ABC):
 
     @abstractmethod
     def get_refs_containing(self, rev: str, pattern: Optional[str] = None):
-        """Iterate over all git refs containing the specfied revision."""
+        """Iterate over all git refs containing the specified revision."""
 
     @abstractmethod
     def push_refspec(

dvc/stage/__init__.py

@@ -4,7 +4,7 @@ import string
 from collections import defaultdict
 from contextlib import suppress
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Dict, Optional, Set
+from typing import TYPE_CHECKING, Any, Dict, Optional, Set
 
 from funcy import cached_property, project
 
@@ -390,6 +390,19 @@ class Stage(params.StageParams):
         if purge:
             self.dvcfile.remove_stage(self)
 
+    def transfer(
+        self,
+        source: str,
+        odb: "ObjectDB" = None,
+        to_remote: bool = False,
+        **kwargs: Any,
+    ) -> None:
+        assert len(self.outs) == 1
+        (out,) = self.outs
+        out.transfer(source, odb=odb, jobs=kwargs.get("jobs"))
+        if not to_remote:
+            out.checkout()
+
     @rwlocked(read=["deps"], write=["outs"])
     def reproduce(self, interactive=False, **kwargs):
         if not (kwargs.get("force", False) or self.changed()):

dvc/ui/__init__.py

@@ -13,6 +13,8 @@ from typing import (
 from funcy import cached_property
 
 if TYPE_CHECKING:
+    from rich.status import Status
+
     from dvc.progress import Tqdm
     from dvc.ui.table import Headers, Styles, TableData
 
@@ -79,18 +81,23 @@ class Console:
         styled: bool = False,
         file: TextIO = None,
     ) -> None:
+        from dvc.progress import Tqdm
+
         sep = " " if sep is None else sep
         end = "\n" if end is None else end
         if not self._enabled and not force:
             return
 
-        if styled:
-            console = self.error_console if stderr else self.rich_console
-            return console.print(*objects, sep=sep, end=end)
-
         file = file or (sys.stderr if stderr else sys.stdout)
-        values = (self.formatter.format(obj, style=style) for obj in objects)
-        return print(*values, sep=sep, end=end, file=file)
+        with Tqdm.external_write_mode(file=file):
+            if styled:
+                console = self.error_console if stderr else self.rich_console
+                return console.print(*objects, sep=sep, end=end)
+
+            values = (
+                self.formatter.format(obj, style=style) for obj in objects
+            )
+            return print(*values, sep=sep, end=end, file=file)
 
     @staticmethod
     def progress(*args, **kwargs) -> "Tqdm":
@@ -181,6 +188,9 @@ class Console:
             self, data, headers, markdown=markdown, pager=pager, force=force
         )
 
+    def status(self, status: str, **kwargs: Any) -> "Status":
+        return self.error_console.status(status, **kwargs)
+
 
 ui = Console()
 

dvc/version.py

@@ -6,7 +6,7 @@
 import os
 import subprocess
 
-_BASE_VERSION = "2.5.0"
+_BASE_VERSION = "2.5.4"
 
 
 def _generate_version(base_version):

scripts/build_package.sh

@@ -11,4 +11,4 @@ fi
 echo 'PKG = "pip"' >dvc/utils/build.py
 
 python setup.py sdist
-python setup.py bdist_wheel --universal
+python setup.py bdist_wheel

scripts/innosetup/setup.iss

@@ -34,7 +34,7 @@ Source: "{#MyAppDir}\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs c
 
 [Tasks]
 Name: modifypath; Description: Adds dvc's application directory to environmental path; Flags: checkablealone;
-Name: modifypath\system; Description: Adds dvc's application directory to enviromental path for all users;
+Name: modifypath\system; Description: Adds dvc's application directory to environmental path for all users;
 Name: addsymlinkpermissions; Description: Add permission for creating symbolic links; Flags: checkablealone;
 Name: addsymlinkpermissions\system; Description: Add permissions for creating symbolic links for all users;
 

scripts/pypi_upload.sh

@@ -9,5 +9,5 @@ fi
 
 rm -rf dist/
 pip install twine
-python setup.py sdist bdist_wheel --universal
+python setup.py sdist bdist_wheel
 twine upload dist/*

setup.cfg

@@ -1,6 +1,3 @@
-[bdist_wheel]
-universal=1
-
 [flake8]
 ignore=
     E203, # Whitespace before ':'

setup.py

@@ -78,6 +78,7 @@ install_requires = [
     "pydot>=1.2.4",
     "speedcopy>=2.0.1; python_version < '3.8' and sys_platform == 'win32'",
     "dataclasses==0.7; python_version < '3.7'",
+    "importlib-metadata>=1.4; python_version < '3.8'",
     "flatten_dict>=0.3.0,<1",
     "tabulate>=0.8.7",
     "pygtrie>=2.3.2",

tests/func/plots/test_show.py

@@ -571,7 +571,7 @@ def test_raise_on_wrong_field(tmp_dir, scm, dvc, run_copy_metrics):
 
 
 def test_load_metric_from_dict_json(tmp_dir):
-    metric = [{"acccuracy": 1, "loss": 2}, {"accuracy": 3, "loss": 4}]
+    metric = [{"accuracy": 1, "loss": 2}, {"accuracy": 3, "loss": 4}]
     dmetric = {"train": metric}
 
     plot_data = JSONPlotData("-", "revision", json.dumps(dmetric))
@@ -584,7 +584,7 @@ def test_load_metric_from_dict_json(tmp_dir):
 
 
 def test_load_metric_from_dict_yaml(tmp_dir):
-    metric = [{"acccuracy": 1, "loss": 2}, {"accuracy": 3, "loss": 4}]
+    metric = [{"accuracy": 1, "loss": 2}, {"accuracy": 3, "loss": 4}]
     dmetric = {"train": metric}
 
     plot_data = YAMLPlotData("-", "revision", dumps_yaml(dmetric))

tests/func/test_add.py

@@ -1,6 +1,5 @@
 import errno
 import filecmp
-import logging
 import os
 import shutil
 import stat
@@ -131,14 +130,12 @@ class TestAddCmdDirectoryRecursive(TestDvc):
 
     def test_warn_about_large_directories(self):
         warning = (
-            "You are adding a large directory 'large-dir' recursively,"
-            " consider tracking it as a whole instead.\n"
-            "{purple}HINT:{nc} Remove the generated DVC file and then"
-            " run `{cyan}dvc add large-dir{nc}`".format(
-                purple=colorama.Fore.MAGENTA,
-                cyan=colorama.Fore.CYAN,
-                nc=colorama.Style.RESET_ALL,
-            )
+            "You are adding a large directory 'large-dir' recursively."
+            "\nConsider tracking it as a whole instead with "
+            "`{cyan}dvc add large-dir{nc}`"
+        ).format(
+            cyan=colorama.Fore.CYAN,
+            nc=colorama.Style.RESET_ALL,
         )
 
         os.mkdir("large-dir")
@@ -149,9 +146,8 @@ class TestAddCmdDirectoryRecursive(TestDvc):
             with open(path, "w") as fobj:
                 fobj.write(path)
 
-        with self._caplog.at_level(logging.WARNING, logger="dvc"):
-            assert main(["add", "--recursive", "large-dir"]) == 0
-            assert warning in self._caplog.messages
+        assert main(["add", "--recursive", "large-dir"]) == 0
+        assert warning in self._capsys.readouterr()[1]
 
 
 class TestAddDirectoryWithForwardSlash(TestDvc):
@@ -643,7 +639,7 @@ def test_failed_add_cleanup(tmp_dir, scm, dvc):
 
 
 def test_should_not_track_git_internal_files(mocker, dvc, tmp_dir):
-    stage_creator_spy = mocker.spy(dvc_module.repo.add, "_create_stages")
+    stage_creator_spy = mocker.spy(dvc_module.repo.add, "create_stages")
 
     ret = main(["add", "-R", dvc.root_dir])
     assert ret == 0
@@ -951,15 +947,15 @@ def test_add_file_in_symlink_dir(make_tmp_dir, tmp_dir, dvc, external):
         dvc.add(os.path.join("dir", "foo"))
 
 
-def test_add_with_cache_link_error(tmp_dir, dvc, mocker, caplog):
+def test_add_with_cache_link_error(tmp_dir, dvc, mocker, capsys):
     tmp_dir.gen("foo", "foo")
 
     mocker.patch(
         "dvc.checkout._do_link", side_effect=DvcException("link failed")
     )
-    with caplog.at_level(logging.WARNING, logger="dvc"):
-        dvc.add("foo")
-        assert "reconfigure cache types" in caplog.text
+    dvc.add("foo")
+    err = capsys.readouterr()[1]
+    assert "reconfigure cache types" in err
 
     assert not (tmp_dir / "foo").exists()
     assert (tmp_dir / "foo.dvc").exists()

tests/func/test_status.py

@@ -49,7 +49,7 @@ def test_status_non_dvc_repo_import(tmp_dir, dvc, git_dir):
 
 
 def test_status_before_and_after_dvc_init(tmp_dir, dvc, git_dir):
-    git_dir.scm_gen("file", "first version", commit="first verison")
+    git_dir.scm_gen("file", "first version", commit="first version")
     old_rev = git_dir.scm.get_rev()
 
     dvc.imp(os.fspath(git_dir), "file", "file")

tests/func/test_version.py

@@ -1,7 +1,7 @@
 import re
 
 from dvc.main import main
-from tests.unit.test_info import PYTHON_VERSION_REGEX
+from tests.unit.test_info import PYTHON_VERSION_REGEX, find_supported_remotes
 
 
 def test_(tmp_dir, dvc, scm, capsys):
@@ -10,7 +10,7 @@ def test_(tmp_dir, dvc, scm, capsys):
     out, _ = capsys.readouterr()
     assert re.search(r"DVC version: \d+\.\d+\.\d+.*", out)
     assert re.search(f"Platform: {PYTHON_VERSION_REGEX} on .*", out)
-    assert re.search(r"Supports: .*", out)
+    assert find_supported_remotes(out)
     assert re.search(r"Cache types: .*", out)
     assert re.search(r"Caches: local", out)
     assert re.search(r"Remotes: None", out)

tests/unit/command/test_diff.py

@@ -48,8 +48,7 @@ def test_default(mocker, capsys):
         "Renamed:\n"
         "    data{sep}file_old -> data{sep}file_new\n"
         "\n"
-        "files summary: 1 added, 0 deleted, "
-        "1 renamed, 0 modified, 0 not in cache"
+        "files summary: 1 added, 1 renamed"
     ).format(sep=os.path.sep) in capsys.readouterr()[0]
 
 
@@ -99,8 +98,7 @@ def test_show_hash(mocker, capsys):
         "    CCCCCCCC..DDDDDDDD  file1\n"
         "    AAAAAAAA..BBBBBBBB  file2\n"
         "\n"
-        "files summary: 0 added, 2 deleted, "
-        "1 renamed, 2 modified, 0 not in cache"
+        "files summary: 2 deleted, 1 renamed, 2 modified"
     ) in out
 
 
@@ -350,6 +348,6 @@ def test_hide_missing(mocker, capsys):
         "Renamed:\n"
         "    file_old -> file_new\n"
         "\n"
-        "files summary: 1 added, 0 deleted, 1 renamed, 0 modified"
+        "files summary: 1 added, 1 renamed"
     ) in out
     assert "not in cache" not in out

tests/unit/objects/test_tree.py

@@ -69,19 +69,27 @@ def test_list(lst, trie_dict):
         ({}, 0),
         (
             {
-                ("a",): HashInfo("md5", "abc", size=1),
-                ("b",): HashInfo("md5", "def", size=2),
-                ("c",): HashInfo("md5", "ghi", size=3),
-                ("dir", "foo"): HashInfo("md5", "jkl", size=4),
-                ("dir", "bar"): HashInfo("md5", "mno", size=5),
-                ("dir", "baz"): HashInfo("md5", "pqr", size=6),
+                ("a",): HashFile(None, None, HashInfo("md5", "abc", size=1)),
+                ("b",): HashFile(None, None, HashInfo("md5", "def", size=2)),
+                ("c",): HashFile(None, None, HashInfo("md5", "ghi", size=3)),
+                ("dir", "foo"): HashFile(
+                    None, None, HashInfo("md5", "jkl", size=4)
+                ),
+                ("dir", "bar"): HashFile(
+                    None, None, HashInfo("md5", "mno", size=5)
+                ),
+                ("dir", "baz"): HashFile(
+                    None, None, HashInfo("md5", "pqr", size=6)
+                ),
             },
             21,
         ),
         (
             {
-                ("a",): HashInfo("md5", "abc", size=1),
-                ("b",): HashInfo("md5", "def", size=None),
+                ("a",): HashFile(None, None, HashInfo("md5", "abc", size=1)),
+                ("b",): HashFile(
+                    None, None, HashInfo("md5", "def", size=None)
+                ),
             },
             None,
         ),
@@ -90,6 +98,7 @@ def test_list(lst, trie_dict):
 def test_size(trie_dict, size):
     tree = Tree(None, None, None)
     tree._dict = trie_dict
+    tree.digest()
     assert tree.size == size
 
 

tests/unit/remote/test_http.py

@@ -175,5 +175,9 @@ def test_content_length(mocker, headers, expected_size):
 
     url = URLInfo("https://example.org/file.txt")
 
-    assert fs.info(url) == {"etag": None, "size": expected_size}
+    assert fs.info(url) == {
+        "etag": None,
+        "size": expected_size,
+        "type": "file",
+    }
     assert fs._content_length(res) == expected_size

tests/unit/test_info.py

@@ -13,6 +13,35 @@ from dvc.info import get_dvc_info
 PYTHON_VERSION_REGEX = r"Python \d\.\d+\.\d+\S*"
 
 
+def find_supported_remotes(string):
+    lines = string.splitlines()
+    index = 0
+
+    for index, line in enumerate(lines):
+        if line == "Supports:":
+            index += 1
+            break
+    else:
+        return []
+
+    remotes = {}
+    for line in lines[index:]:
+        if not line.startswith("\t"):
+            break
+
+        remote_name, _, raw_dependencies = (
+            line.strip().strip(",").partition(" ")
+        )
+        remotes[remote_name] = {
+            dependency: version
+            for dependency, _, version in [
+                dependency.partition(" = ")
+                for dependency in raw_dependencies[1:-1].split(", ")
+            ]
+        }
+    return remotes
+
+
 @pytest.mark.parametrize("scm_init", [True, False])
 def test_info_in_repo(scm_init, tmp_dir):
     tmp_dir.init(scm=scm_init, dvc=True)
@@ -23,7 +52,7 @@ def test_info_in_repo(scm_init, tmp_dir):
 
     assert re.search(r"DVC version: \d+\.\d+\.\d+.*", dvc_info)
     assert re.search(f"Platform: {PYTHON_VERSION_REGEX} on .*", dvc_info)
-    assert re.search(r"Supports: .*", dvc_info)
+    assert find_supported_remotes(dvc_info)
     assert re.search(r"Cache types: .*", dvc_info)
 
     if scm_init:
@@ -98,7 +127,7 @@ def test_info_outside_of_repo(tmp_dir, caplog):
 
     assert re.search(r"DVC version: \d+\.\d+\.\d+.*", dvc_info)
     assert re.search(f"Platform: {PYTHON_VERSION_REGEX} on .*", dvc_info)
-    assert re.search(r"Supports: .*", dvc_info)
+    assert find_supported_remotes(dvc_info)
     assert not re.search(r"Cache types: .*", dvc_info)
     assert "Repo:" not in dvc_info
 
@@ -107,4 +136,14 @@ def test_fs_info_outside_of_repo(tmp_dir, caplog):
     dvc_info = get_dvc_info()
     assert re.search(r"DVC version: \d+\.\d+\.\d+.*", dvc_info)
     assert re.search(f"Platform: {PYTHON_VERSION_REGEX} on .*", dvc_info)
-    assert re.search(r"Supports: .*", dvc_info)
+    assert find_supported_remotes(dvc_info)
+
+
+def test_plugin_versions(tmp_dir, dvc):
+    from dvc.fs import FS_MAP
+
+    dvc_info = get_dvc_info()
+    remotes = find_supported_remotes(dvc_info)
+
+    for remote, dependencies in remotes.items():
+        assert dependencies.keys() == FS_MAP[remote].REQUIRES.keys()