Custom header for dvc webdav (#8933)Fix https://github.com/iterative/dvc/issues/8931

* include custom auth header for webdav

* Revert "include custom auth header for webdav"

This reverts commit d2f2b2889b6bcf0429e4a28076f709a04642ae3a.

* include custom auth header for dvc webdav, ran unit tests

* update dvc-webdav version

.flake8

@@ -13,6 +13,7 @@ ignore=
     B028
 max_line_length = 79
 max-complexity = 10
-select = B,C,E,F,W,T4,B902,T,P
+select = B,C,E,F,W,T4,B902,T,P,TYP
 show_source = true
 count = true
+min_python_version = 3.8.0

.github/workflows/plugin_tests.yaml

@@ -0,0 +1,54 @@
+name: Remote Plugin Tests
+
+on:
+  pull_request:
+  workflow_dispatch:
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  tests:
+    timeout-minutes: 45
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-20.04]
+        pyv: ["3.10"]
+        plugin: ["dvc-s3"]
+
+    steps:
+    - uses: actions/checkout@v3
+      with:
+        path: dvc
+
+    - uses: actions/checkout@v3
+      with:
+        repository: iterative/${{ matrix.plugin }}
+        ref: main
+        path: ${{ matrix.plugin }}
+
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.pyv }}
+        cache: pip
+        cache-dependency-path: |
+          dvc/pyproject.toml
+          ${{ matrix.plugin }}/setup.cfg
+
+    - name: Install plugin + DVC@PR
+      run: |
+        pip install --upgrade pip wheel
+        pip install "./dvc[testing]"
+        pip install -e "./${{ matrix.plugin }}[tests]"
+
+    - name: Run plugin tests
+      timeout-minutes: 15
+      working-directory: ${{ matrix.plugin }}
+      run: pytest -v -n=auto

.pre-commit-config.yaml

@@ -39,14 +39,13 @@ repos:
     hooks:
       - id: black
   - repo: https://github.com/PyCQA/isort
-    rev: 5.11.4
+    rev: 5.12.0
     hooks:
       - id: isort
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: 'v0.0.223'
+    rev: 'v0.0.238'
     hooks:
       - id: ruff
-        args: ["--force-exclude"]
   - repo: https://github.com/codespell-project/codespell
     rev: v2.2.2
     hooks:
@@ -61,6 +60,7 @@ repos:
           - flake8-comprehensions
           - flake8-debugger
           - flake8-string-format
+          - flake8-typing-imports
   - repo: local
     hooks:
     - id: mypy

dvc/analytics.py

@@ -82,10 +82,9 @@ def send(path):
 
 
 def _scm_in_use():
-    from scmrepo.noscm import NoSCM
-
     from dvc.exceptions import NotDvcRepoError
     from dvc.repo import Repo
+    from dvc.scm import NoSCM
 
     from .scm import SCM, SCMError
 
@@ -102,17 +101,24 @@ def _runtime_info():
     """
     Gather information from the environment where DVC runs to fill a report.
     """
-    from iterative_telemetry import find_or_create_user_id
+    from iterative_telemetry import _generate_ci_id, find_or_create_user_id
 
     from dvc import __version__
     from dvc.utils import is_binary
 
+    ci_id = _generate_ci_id()
+    if ci_id:
+        group_id, user_id = ci_id
+    else:
+        group_id, user_id = None, find_or_create_user_id()
+
     return {
         "dvc_version": __version__,
         "is_binary": is_binary(),
         "scm_class": _scm_in_use(),
         "system_info": _system_info(),
-        "user_id": find_or_create_user_id(),
+        "user_id": user_id,
+        "group_id": group_id,
     }
 
 

dvc/annotations.py

@@ -12,7 +12,7 @@ class Annotation:
     PARAM_META: ClassVar[str] = "meta"
 
     desc: Optional[str] = None
-    type: Optional[str] = None
+    type: Optional[str] = None  # noqa: A003
     labels: List[str] = field(default_factory=list)
     meta: Dict[str, Any] = field(default_factory=dict)
 

dvc/cli/__init__.py

@@ -57,8 +57,8 @@ def _log_exceptions(exc: Exception) -> Optional[int]:
         if exc.errno == errno.EMFILE:
             logger.exception(
                 "too many open files, please visit "
-                "{} to see how to handle this "
-                "problem".format(error_link("many-files")),
+                "%s to see how to handle this problem",
+                error_link("many-files"),
                 extra={"tb_only": True},
             )
         else:
@@ -91,8 +91,11 @@ def _log_exceptions(exc: Exception) -> Optional[int]:
             hint = f"\nPlease report this bug to {link}. Thank you!"
 
         logger.exception(
-            f"URL '{exc.url}' is supported but requires these missing "
-            f"dependencies: {exc.missing_deps}. {hint}",
+            "URL '%s' is supported but requires these missing "
+            "dependencies: %s. %s",
+            exc.url,
+            exc.missing_deps,
+            hint,
             extra={"tb_only": True},
         )
         return None
@@ -101,7 +104,9 @@ def _log_exceptions(exc: Exception) -> Optional[int]:
         link = format_link("https://man.dvc.org/remote/modify")
         logger.exception("configuration error")
         logger.exception(
-            f"{exc!s}\nLearn more about configuration settings at {link}.",
+            "%s\nLearn more about configuration settings at %s.",
+            exc,
+            link,
             extra={"tb_only": True},
         )
         return 251
@@ -113,9 +118,12 @@ def _log_exceptions(exc: Exception) -> Optional[int]:
 
         directory = relpath(exc.directory)
         logger.exception(
-            f"Could not open pickled '{exc.type}' cache.\n"
-            f"Remove the '{directory}' directory and then retry this command."
-            f"\nSee {error_link('pickle')} for more information.",
+            "Could not open pickled '%s' cache.\n"
+            "Remove the '%s' directory and then retry this command."
+            "\nSee %s for more information.",
+            exc.type,
+            directory,
+            error_link("pickle"),
             extra={"tb_only": True},
         )
         return None
@@ -172,6 +180,21 @@ def main(argv=None):  # noqa: C901
         if level is not None:
             set_loggers_level(level)
 
+        if level and level <= logging.DEBUG:
+            from platform import (
+                platform,
+                python_implementation,
+                python_version,
+            )
+
+            from dvc import __version__
+            from dvc.utils.pkg import PKG
+
+            pyv = " ".join([python_implementation(), python_version()])
+            pkg = f" ({PKG})" if PKG else ""
+            logger.debug("v%s%s, %s on %s", __version__, pkg, pyv, platform())
+            logger.debug("command: %s", " ".join(argv or sys.argv))
+
         logger.trace(args)  # type: ignore[attr-defined]
 
         if not sys.stdout.closed and not args.quiet:

dvc/cli/actions.py

@@ -2,7 +2,9 @@ from argparse import _AppendAction
 
 
 class CommaSeparatedArgs(_AppendAction):  # pylint: disable=protected-access
-    def __call__(self, parser, namespace, values, option_string=None):
+    def __call__(
+        self, parser, namespace, values, option_string=None  # noqa: ARG002
+    ):
         from funcy import ldistinct
 
         items = getattr(namespace, self.dest) or []
@@ -11,7 +13,9 @@ class CommaSeparatedArgs(_AppendAction):  # pylint: disable=protected-access
 
 
 class KeyValueArgs(_AppendAction):
-    def __call__(self, parser, namespace, values, option_string=None):
+    def __call__(
+        self, parser, namespace, values, option_string=None  # noqa: ARG002
+    ):
         items = getattr(namespace, self.dest) or {}
         for value in filter(bool, values):
             key, _, value = value.partition("=")

dvc/commands/commit.py

@@ -25,9 +25,7 @@ class CmdCommit(CmdBase):
                 )
             except DvcException:
                 logger.exception(
-                    "failed to commit{}".format(
-                        (" " + target) if target else ""
-                    )
+                    "failed to commit%s", (" " + target) if target else ""
                 )
                 return 1
         return 0

dvc/commands/experiments/show.py

@@ -77,7 +77,7 @@ def _collect_rows(  # noqa: C901
     fill_value=FILL_VALUE,
     iso=False,
 ):
-    from scmrepo.git import Git
+    from dvc.scm import Git
 
     if sort_by:
         sort_path, sort_name, sort_type = _sort_column(

dvc/commands/get.py

@@ -45,13 +45,11 @@ class CmdGet(CmdBaseNoRepo):
             )
             return 0
         except CloneError:
-            logger.exception(f"failed to get '{self.args.path}'")
+            logger.exception("failed to get '%s'", self.args.path)
             return 1
         except DvcException:
             logger.exception(
-                "failed to get '{}' from '{}'".format(
-                    self.args.path, self.args.url
-                )
+                "failed to get '%s' from '%s'", self.args.path, self.args.url
             )
             return 1
 

dvc/commands/imp.py

@@ -28,13 +28,13 @@ class CmdImport(CmdBase):
                 jobs=self.args.jobs,
             )
         except CloneError:
-            logger.exception(f"failed to import '{self.args.path}'")
+            logger.exception("failed to import '%s'", self.args.path)
             return 1
         except DvcException:
             logger.exception(
-                "failed to import '{}' from '{}'.".format(
-                    self.args.path, self.args.url
-                )
+                "failed to import '%s' from '%s'.",
+                self.args.path,
+                self.args.url,
             )
             return 1
         return 0

dvc/commands/imp_url.py

@@ -29,10 +29,9 @@ class CmdImportUrl(CmdBase):
             )
         except DvcException:
             logger.exception(
-                "failed to import {}. You could also try downloading "
-                "it manually, and adding it with `dvc add`.".format(
-                    self.args.url
-                )
+                "failed to import %s. You could also try downloading "
+                "it manually, and adding it with `dvc add`.",
+                self.args.url,
             )
             return 1
         return 0

dvc/commands/ls/ls_colors.py

@@ -20,7 +20,7 @@ class LsColors:
             else:
                 self._codes[code] = color
 
-    def format(self, entry):
+    def format(self, entry):  # noqa: A003
         text = entry["path"]
 
         if entry.get("isout", False) and "out" in self._codes:

dvc/compare.py

@@ -329,8 +329,8 @@ def _format_field(
     def _format(_val):
         if isinstance(_val, float) and precision:
             if round_digits:
-                return round(val, precision)
-            return _normalize_float(val, precision)
+                return round(_val, precision)
+            return _normalize_float(_val, precision)
         if isinstance(_val, abc.Mapping):
             return {k: _format(v) for k, v in _val.items()}
         if isinstance(_val, list):

dvc/config.py

@@ -9,12 +9,12 @@ from typing import TYPE_CHECKING, Dict, Optional
 from funcy import compact, memoize, re_find
 
 from dvc.exceptions import DvcException, NotDvcRepoError
-from dvc.types import DictStrAny, StrPath
 
 from .utils.objects import cached_property
 
 if TYPE_CHECKING:
     from dvc.fs import FileSystem
+    from dvc.types import DictStrAny, StrPath
 
 logger = logging.getLogger(__name__)
 
@@ -92,10 +92,10 @@ class Config(dict):
 
     def __init__(
         self,
-        dvc_dir: Optional[StrPath] = None,
+        dvc_dir: Optional["StrPath"] = None,
         validate: bool = True,
         fs: Optional["FileSystem"] = None,
-        config: Optional[DictStrAny] = None,
+        config: Optional["DictStrAny"] = None,
     ):  # pylint: disable=super-init-not-called
         from dvc.fs import LocalFileSystem
 
@@ -153,7 +153,9 @@ class Config(dict):
         with open(config_file, "w+", encoding="utf-8"):
             return Config(dvc_dir)
 
-    def load(self, validate: bool = True, config: Optional[DictStrAny] = None):
+    def load(
+        self, validate: bool = True, config: Optional["DictStrAny"] = None
+    ):
         """Loads config from all the config files.
 
         Raises:

dvc/config_schema.py

@@ -106,6 +106,7 @@ WEBDAV_COMMON = {
     "password": str,
     "ask_password": Bool,
     "token": str,
+    "custom_auth_header": str,
     "cert_path": str,
     "key_path": str,
     "timeout": Coerce(int),

dvc/daemon.py

@@ -105,7 +105,7 @@ def daemon(args):
     Args:
         args (list): list of arguments to append to `dvc daemon` command.
     """
-    daemonize(["daemon", "-q"] + args)
+    daemonize(["daemon", "-q", *args])
 
 
 def daemonize(cmd: List[str]):

dvc/data_cloud.py

@@ -8,6 +8,7 @@ from dvc.utils.objects import cached_property
 from dvc_data.hashfile.db import get_index
 
 if TYPE_CHECKING:
+    from dvc.fs import FileSystem
     from dvc_data.hashfile.db import HashFileDB
     from dvc_data.hashfile.hash_info import HashInfo
     from dvc_data.hashfile.status import CompareStatusResult
@@ -17,20 +18,13 @@ logger = logging.getLogger(__name__)
 
 
 class Remote:
-    def __init__(self, path, fs, **config):
+    def __init__(self, name: str, path: str, fs: "FileSystem", **config):
         self.path = path
         self.fs = fs
+        self.name = name
 
-        self.worktree = config.pop("worktree", False)
+        self.worktree: bool = config.pop("worktree", False)
         self.config = config
-        if self.worktree:
-            version_aware = self.config.get("version_aware")
-            if version_aware is False:
-                raise RemoteConfigError(
-                    "worktree remotes require version_aware cloud"
-                )
-            if version_aware is None:
-                self.fs.version_aware = True
 
     @cached_property
     def odb(self) -> "HashFileDB":
@@ -69,9 +63,19 @@ class DataCloud:
             from dvc.fs import get_cloud_fs
 
             cls, config, fs_path = get_cloud_fs(self.repo, name=name)
+
+            if config.get("worktree"):
+                version_aware = config.get("version_aware")
+                if version_aware is False:
+                    raise RemoteConfigError(
+                        "worktree remotes require version_aware cloud"
+                    )
+                if version_aware is None:
+                    config["version_aware"] = True
+
             fs = cls(**config)
             config["tmp_dir"] = self.repo.index_db_dir
-            return Remote(fs_path, fs, **config)
+            return Remote(name, fs_path, fs, **config)
 
         if bool(self.repo.config["remote"]):
             error_msg = (
@@ -108,7 +112,8 @@ class DataCloud:
             )
             logger.warning(
                 "Some of the cache files do not exist neither locally "
-                f"nor on remote. Missing cache files:\n{missing_desc}"
+                "nor on remote. Missing cache files:\n%s",
+                missing_desc,
             )
 
     def transfer(

dvc/dependency/__init__.py

@@ -74,7 +74,7 @@ def _merge_params(s_list):
         if not isinstance(key, dict):
             msg = "Only list of str/dict is supported. Got: "
             msg += f"'{type(key).__name__}'."
-            raise ValueError(msg)
+            raise ValueError(msg)  # noqa: TRY004
 
         for k, params in key.items():
             if k in wholly_tracked:
@@ -83,7 +83,7 @@ def _merge_params(s_list):
             if not isinstance(params, list):
                 msg = "Expected list of params for custom params file "
                 msg += f"'{k}', got '{type(params).__name__}'."
-                raise ValueError(msg)
+                raise ValueError(msg)  # noqa: TRY004
             d[k].extend(params)
     return d
 

dvc/dependency/param.py

@@ -130,8 +130,7 @@ class ParamsDependency(Dependency):
                     and list(actual[param]) == info[param]
                 ):
                     continue
-                else:
-                    st = "modified"
+                st = "modified"
             else:
                 continue
 

dvc/dependency/repo.py

@@ -1,28 +1,22 @@
 import os
 from collections import defaultdict
 from copy import copy
-from typing import (
-    TYPE_CHECKING,
-    ContextManager,
-    Dict,
-    Optional,
-    Set,
-    Tuple,
-    Union,
-)
+from typing import TYPE_CHECKING, Dict, Optional, Set, Tuple, Union
 
 from voluptuous import Required
 
 from dvc.prompt import confirm
-from dvc_data.hashfile.meta import Meta
 
 from .base import Dependency
 
 if TYPE_CHECKING:
+    from typing import ContextManager
+
     from dvc.output import Output
     from dvc.repo import Repo
     from dvc.stage import Stage
     from dvc_data.hashfile.hash_info import HashInfo
+    from dvc_data.hashfile.meta import Meta
     from dvc_data.hashfile.obj import HashFile
     from dvc_objects.db import ObjectDB
 
@@ -46,10 +40,10 @@ class RepoDependency(Dependency):
     ):
         self.def_repo = def_repo
         self._objs: Dict[str, "HashFile"] = {}
-        self._meta: Dict[str, Meta] = {}
+        self._meta: Dict[str, "Meta"] = {}
         super().__init__(stage, *args, **kwargs)
 
-    def _parse_path(self, fs, fs_path):
+    def _parse_path(self, fs, fs_path):  # noqa: ARG002
         return None
 
     @property
@@ -114,7 +108,9 @@ class RepoDependency(Dependency):
 
     def _get_used_and_obj(
         self, obj_only: bool = False, **kwargs
-    ) -> Tuple[Dict[Optional["ObjectDB"], Set["HashInfo"]], Meta, "HashFile"]:
+    ) -> Tuple[
+        Dict[Optional["ObjectDB"], Set["HashInfo"]], "Meta", "HashFile"
+    ]:
         from dvc.config import NoRemoteError
         from dvc.exceptions import NoOutputOrStageError, PathMissingError
         from dvc.utils import as_posix
@@ -143,7 +139,7 @@ class RepoDependency(Dependency):
                         self._check_circular_import(odb, obj_ids)
                         used_obj_ids[odb].update(obj_ids)
                 except (NoRemoteError, NoOutputOrStageError):
-                    pass
+                    pass  # noqa: S110
 
             try:
                 object_store, meta, obj = build(

dvc/dvcfile.py

@@ -21,7 +21,6 @@ from dvc.stage.exceptions import (
     StageFileDoesNotExistError,
     StageFileIsNotDvcFileError,
 )
-from dvc.types import StrOrBytesPath
 from dvc.utils import relpath
 from dvc.utils.collections import apply_diff
 from dvc.utils.objects import cached_property
@@ -29,6 +28,7 @@ from dvc.utils.serialize import dump_yaml, modify_yaml
 
 if TYPE_CHECKING:
     from dvc.repo import Repo
+    from dvc.types import StrOrBytesPath
 
     from .parsing import DataResolver
     from .stage import Stage
@@ -179,7 +179,8 @@ class FileMixin:
             **kwargs,
         )
 
-    def remove(self, force=False):  # pylint: disable=unused-argument
+    # pylint: disable-next=unused-argument
+    def remove(self, force=False):  # noqa: ARG002
         with contextlib.suppress(FileNotFoundError):
             os.unlink(self.path)
 
@@ -219,7 +220,8 @@ class SingleStageFile(FileMixin):
         dump_yaml(self.path, serialize.to_single_stage_file(stage, **kwargs))
         self.repo.scm_context.track_file(self.relpath)
 
-    def remove_stage(self, stage):  # pylint: disable=unused-argument
+    # pylint: disable-next=unused-argument
+    def remove_stage(self, stage):  # noqa: ARG002
         self.remove()
 
     def merge(self, ancestor, other, allowed=None):
@@ -465,7 +467,7 @@ class Lockfile(FileMixin):
 
 
 def load_file(
-    repo: "Repo", path: StrOrBytesPath, **kwargs: Any
+    repo: "Repo", path: "StrOrBytesPath", **kwargs: Any
 ) -> Union[ProjectFile, SingleStageFile]:
     _, ext = os.path.splitext(path)
     if ext in (".yaml", ".yml"):

dvc/external_repo.py

@@ -17,11 +17,12 @@ from dvc.exceptions import (
 )
 from dvc.repo import Repo
 from dvc.scm import CloneError, map_scm_exception
-from dvc.types import StrPath
 from dvc.utils import relpath
 
 if TYPE_CHECKING:
-    from scmrepo import Git
+    from dvc.types import StrPath
+
+    from .scm import Git
 
 logger = logging.getLogger(__name__)
 
@@ -32,14 +33,13 @@ def external_repo(
     url,
     rev: Optional[str] = None,
     for_write: bool = False,
-    cache_dir: Optional[StrPath] = None,
+    cache_dir: Optional["StrPath"] = None,
     cache_types: Optional[Collection[str]] = None,
     **kwargs,
 ) -> Iterator["Repo"]:
-    from scmrepo.git import Git
-
     from dvc.config import NoRemoteError
     from dvc.fs import GitFileSystem
+    from dvc.scm import Git
 
     logger.debug("Creating external repo %s@%s", url, rev)
     path = _cached_clone(url, rev, for_write=for_write)
@@ -196,7 +196,7 @@ def _clone_default_branch(url, rev, for_write=False):  # noqa: C901
 
     The cloned is reactualized with git pull unless rev is a known sha.
     """
-    from scmrepo.git import Git
+    from dvc.scm import Git
 
     clone_path, shallow = CLONES.get(url) or (None, False)
 
@@ -273,7 +273,7 @@ def _merge_upstream(git: "Git"):
 
 
 def _git_checkout(repo_path, rev):
-    from scmrepo.git import Git
+    from dvc.scm import Git
 
     logger.debug("erepo: git checkout %s@%s", repo_path, rev)
     git = Git(repo_path)

dvc/fs/callbacks.py

@@ -70,7 +70,7 @@ class RichCallback(Callback):
             self.progress.clear_task(self.task)
         self._stack.close()
 
-    def call(self, hook_name=None, **kwargs):
+    def call(self, hook_name=None, **kwargs):  # noqa: ARG002
         self.progress.update(
             self.task,
             completed=self.value,

dvc/fs/dvc.py

@@ -20,7 +20,6 @@ from typing import (
 from fsspec.spec import AbstractFileSystem
 from funcy import wrap_with
 
-from dvc.types import StrPath
 from dvc_objects.fs.base import FileSystem
 from dvc_objects.fs.path import Path
 
@@ -28,6 +27,7 @@ from .data import DataFileSystem
 
 if TYPE_CHECKING:
     from dvc.repo import Repo
+    from dvc.types import StrPath
 
 logger = logging.getLogger(__name__)
 
@@ -166,7 +166,7 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         if hasattr(repo, "dvc_dir"):
             self._datafss[key] = DataFileSystem(index=repo.index.data["repo"])
 
-    def _get_key(self, path: StrPath) -> Key:
+    def _get_key(self, path: "StrPath") -> Key:
         parts = self.repo.fs.path.relparts(path, self.repo.root_dir)
         if parts == (os.curdir,):
             return ()
@@ -304,7 +304,7 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
                 ):
                     fs_infos[fs.path.name(info["name"])] = info
             except (FileNotFoundError, NotADirectoryError):
-                pass
+                pass  # noqa: S110
 
         dvcfiles = kwargs.get("dvcfiles", False)
 

dvc/fs/git.py

@@ -5,9 +5,10 @@ from . import FileSystem
 
 if TYPE_CHECKING:
     from scmrepo.fs import GitFileSystem as FsspecGitFileSystem
-    from scmrepo.git import Git
     from scmrepo.git.objects import GitTrie
 
+    from dvc.scm import Git
+
 
 class GitFileSystem(FileSystem):  # pylint:disable=abstract-method
     """Proxies the repo file access methods to Git objects"""

dvc/ignore.py

@@ -3,15 +3,18 @@ import os
 import re
 from collections import namedtuple
 from itertools import chain, groupby, takewhile
-from typing import List, Optional
+from typing import TYPE_CHECKING, List, Optional
 
 from pathspec.patterns import GitWildMatchPattern
 from pathspec.util import normalize_file
 from pygtrie import Trie
 
-from dvc.fs import AnyFSPath, FileSystem, Schemes, localfs
+from dvc.fs import Schemes, localfs
 from dvc.pathspec_math import PatternInfo, merge_patterns
 
+if TYPE_CHECKING:
+    from dvc.fs import AnyFSPath, FileSystem
+
 logger = logging.getLogger(__name__)
 
 
@@ -306,7 +309,7 @@ class DvcIgnoreFilter:
 
         return [fs_dict[name] for name in chain(dirs, nondirs)]
 
-    def walk(self, fs: FileSystem, path: AnyFSPath, **kwargs):
+    def walk(self, fs: "FileSystem", path: "AnyFSPath", **kwargs):
         detail = kwargs.get("detail", False)
         ignore_subrepos = kwargs.pop("ignore_subrepos", True)
         if fs.protocol == Schemes.LOCAL:
@@ -330,7 +333,7 @@ class DvcIgnoreFilter:
         else:
             yield from fs.walk(path, **kwargs)
 
-    def find(self, fs: FileSystem, path: AnyFSPath, **kwargs):
+    def find(self, fs: "FileSystem", path: "AnyFSPath", **kwargs):
         if fs.protocol == Schemes.LOCAL:
             for root, _, files in self.walk(fs, path, **kwargs):
                 for file in files:
@@ -420,7 +423,7 @@ class DvcIgnoreFilter:
         return _no_match(target)
 
     def is_ignored(
-        self, fs: FileSystem, path: str, ignore_subrepos: bool = True
+        self, fs: "FileSystem", path: str, ignore_subrepos: bool = True
     ) -> bool:
         # NOTE: can't use self.check_ignore(path).match for now, see
         # https://github.com/iterative/dvc/issues/4555

dvc/info.py

@@ -22,13 +22,14 @@ SUBPROJECTS = (
     "dvclive",
     "scmrepo",
 )
-package = "" if PKG is None else f"({PKG})"
+package = "" if PKG is None else f" ({PKG})"
 
 
 def get_dvc_info():
+    dvc_version = f"DVC version: {__version__}{package}"
     info = [
-        f"DVC version: {__version__} {package}",
-        "---------------------------------",
+        dvc_version,
+        "-" * len(dvc_version),
         f"Platform: Python {platform.python_version()} on "
         f"{platform.platform()}",
         f"Subprojects:{_get_subprojects()}",

dvc/logger.py

@@ -68,7 +68,7 @@ class LoggingException(Exception):
 
 def excludeFilter(level):
     class ExcludeLevelFilter(logging.Filter):
-        def filter(self, record):
+        def filter(self, record):  # noqa: A003
             return record.levelno < level
 
     return ExcludeLevelFilter
@@ -94,7 +94,7 @@ class ColorFormatter(logging.Formatter):
         "CRITICAL": colorama.Fore.RED,
     }
 
-    def format(self, record):
+    def format(self, record):  # noqa: A003
         record.message = record.getMessage()
         msg = self.formatMessage(record)
 
@@ -156,7 +156,7 @@ class LoggerHandler(logging.StreamHandler):
                         if not _is_verbose():
                             return
                     except Exception:  # noqa, pylint: disable=broad-except
-                        pass
+                        pass  # noqa: S110
 
             msg = self.format(record)
             Tqdm.write(

dvc/machine/__init__.py

@@ -13,10 +13,10 @@ from typing import (
 )
 
 from dvc.exceptions import DvcException
-from dvc.types import StrPath
 
 if TYPE_CHECKING:
     from dvc.repo import Repo
+    from dvc.types import StrPath
 
     from .backend.base import BaseMachineBackend
 
@@ -76,7 +76,7 @@ class MachineBackends(Mapping):
     def __init__(
         self,
         selected: Optional[Iterable[str]],
-        tmp_dir: StrPath,
+        tmp_dir: "StrPath",
         **kwargs,
     ) -> None:
         selected = selected or list(self.DEFAULT)

dvc/output.py

@@ -4,8 +4,8 @@ from collections import defaultdict
 from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple, Type, cast
 from urllib.parse import urlparse
 
-from funcy import collecting, project
-from voluptuous import And, Any, Coerce, Length, Lower, Required, SetTo
+from funcy import collecting, first, project
+from voluptuous import All, And, Any, Coerce, Length, Lower, Required, SetTo
 
 from dvc import prompt
 from dvc.exceptions import (
@@ -32,7 +32,7 @@ from dvc_objects.errors import ObjectFormatError
 
 from .annotations import ANNOTATION_FIELDS, ANNOTATION_SCHEMA, Annotation
 from .fs import LocalFileSystem, RemoteMissingDepsError, Schemes, get_cloud_fs
-from .fs.callbacks import DEFAULT_CALLBACK, Callback
+from .fs.callbacks import DEFAULT_CALLBACK
 from .utils import relpath
 from .utils.fs import path_isin
 
@@ -41,6 +41,7 @@ if TYPE_CHECKING:
     from dvc_data.index import DataIndexKey
     from dvc_objects.db import ObjectDB
 
+    from .fs.callbacks import Callback
     from .ignore import DvcIgnoreFilter
 
 logger = logging.getLogger(__name__)
@@ -150,11 +151,13 @@ def _merge_data(s_list):
             d[key].update({})
             continue
         if not isinstance(key, dict):
-            raise ValueError(f"'{type(key).__name__}' not supported.")
+            raise ValueError(  # noqa: TRY004
+                f"'{type(key).__name__}' not supported."
+            )
 
         for k, flags in key.items():
             if not isinstance(flags, dict):
-                raise ValueError(
+                raise ValueError(  # noqa: TRY004
                     f"Expected dict for '{k}', got: '{type(flags).__name__}'"
                 )
             d[k].update(flags)
@@ -204,6 +207,30 @@ def load_from_pipeline(stage, data, typ="outs"):
         )
 
 
+def split_file_meta_from_cloud(entry: Dict) -> Dict:
+    if remote_name := entry.pop(Meta.PARAM_REMOTE, None):
+        remote_meta = {}
+        for key in (
+            S3_PARAM_CHECKSUM,
+            HDFS_PARAM_CHECKSUM,
+            Meta.PARAM_VERSION_ID,
+        ):
+            if value := entry.pop(key, None):
+                remote_meta[key] = value
+
+        if remote_meta:
+            entry[Output.PARAM_CLOUD] = {remote_name: remote_meta}
+    return entry
+
+
+def merge_file_meta_from_cloud(entry: Dict) -> Dict:
+    cloud_meta = entry.pop(Output.PARAM_CLOUD, {})
+    if remote_name := first(cloud_meta):
+        entry.update(cloud_meta[remote_name])
+        entry[Meta.PARAM_REMOTE] = remote_name
+    return entry
+
+
 class OutputDoesNotExistError(DvcException):
     def __init__(self, path):
         msg = f"output '{path}' does not exist"
@@ -258,6 +285,7 @@ class Output:
     PARAM_PERSIST = "persist"
     PARAM_REMOTE = "remote"
     PARAM_PUSH = "push"
+    PARAM_CLOUD = "cloud"
 
     METRIC_SCHEMA = Any(
         None,
@@ -297,10 +325,13 @@ class Output:
             desc=desc, type=type, labels=labels or [], meta=meta or {}
         )
         self.repo = stage.repo if not repo and stage else repo
-        meta = Meta.from_dict(info or {})
+        meta_d = merge_file_meta_from_cloud(info or {})
+        meta = Meta.from_dict(meta_d)
         # NOTE: when version_aware is not passed into get_cloud_fs, it will be
         # set based on whether or not path is versioned
-        fs_kwargs = {"version_aware": True} if meta.version_id else {}
+        fs_kwargs = {}
+        if meta.version_id or files:
+            fs_kwargs["version_aware"] = True
 
         if fs_config is not None:
             fs_kwargs.update(**fs_config)
@@ -340,6 +371,9 @@ class Output:
         # should be absolute and don't contain remote:// refs.
         self.stage = stage
         self.meta = meta
+
+        if files is not None:
+            files = [merge_file_meta_from_cloud(f) for f in files]
         self.files = files
         self.use_cache = False if self.IS_DEPENDENCY else cache
         self.metric = False if self.IS_DEPENDENCY else metric
@@ -368,7 +402,18 @@ class Output:
             name=self.hash_name,
             value=getattr(self.meta, self.hash_name, None),
         )
-        if self.meta.nfiles or self.hash_info and self.hash_info.isdir:
+        self._compute_meta_hash_info_from_files()
+
+    def _compute_meta_hash_info_from_files(self) -> None:
+        if self.files:
+            tree = Tree.from_list(self.files, hash_name=self.hash_name)
+            tree.digest()
+            self.hash_info = tree.hash_info
+            self.meta.isdir = True
+            self.meta.nfiles = len(self.files)
+            self.meta.size = sum(f.get("size") for f in self.files)
+            self.meta.remote = first(f.get("remote") for f in self.files)
+        elif self.meta.nfiles or self.hash_info and self.hash_info.isdir:
             self.meta.isdir = True
             if not self.hash_info and self.hash_name != "md5":
                 md5 = getattr(self.meta, "md5", None)
@@ -510,13 +555,6 @@ class Output:
         return workspace, key
 
     def get_entry(self) -> "DataIndexEntry":
-        from dvc.config import NoRemoteError
-
-        try:
-            remote = self.repo.cloud.get_remote_odb(self.remote)
-        except NoRemoteError:
-            remote = None
-
         if self.files and not self.obj:
             self.obj = self.get_obj()
 
@@ -524,9 +562,6 @@ class Output:
             meta=self.meta,
             obj=self.obj,
             hash_info=self.hash_info,
-            odb=self.odb,
-            cache=self.odb,
-            remote=remote,
         )
         if self.stage.is_import and not self.stage.is_repo_import:
             dep = self.stage.deps[0]
@@ -745,9 +780,18 @@ class Output:
         return checkout_obj
 
     def dumpd(self, **kwargs):  # noqa: C901
-        meta = self.meta.to_dict()
-        meta.pop("isdir", None)
-        ret: Dict[str, Any] = {**self.hash_info.to_dict(), **meta}
+        ret: Dict[str, Any] = {}
+        with_files = (
+            (not self.IS_DEPENDENCY or self.stage.is_import)
+            and self.hash_info.isdir
+            and (kwargs.get("with_files") or self.files is not None)
+        )
+
+        if not with_files:
+            meta_d = self.meta.to_dict()
+            meta_d.pop("isdir", None)
+            ret.update(self.hash_info.to_dict())
+            ret.update(split_file_meta_from_cloud(meta_d))
 
         if self.is_in_repo:
             path = self.fs.path.as_posix(
@@ -788,20 +832,14 @@ class Output:
             if not self.can_push:
                 ret[self.PARAM_PUSH] = self.can_push
 
-        if (
-            (not self.IS_DEPENDENCY or self.stage.is_import)
-            and self.hash_info.isdir
-            and (kwargs.get("with_files") or self.files is not None)
-        ):
-            obj: Optional["HashFile"]
-            if self.obj:
-                obj = self.obj
-            else:
-                obj = self.get_obj()
+        if with_files:
+            obj = self.obj or self.get_obj()
             if obj:
-                obj = cast(Tree, obj)
-                ret[self.PARAM_FILES] = obj.as_list(with_meta=True)
-
+                obj = cast("Tree", obj)
+                ret[self.PARAM_FILES] = [
+                    split_file_meta_from_cloud(f)
+                    for f in obj.as_list(with_meta=True)
+                ]
         return ret
 
     def verify_metric(self):
@@ -1226,7 +1264,8 @@ class Output:
         if not self.obj or not other.hash_info.isdir:
             return
         other_obj = other.obj if other.obj is not None else other.get_obj()
-        assert isinstance(self.obj, Tree) and isinstance(other_obj, Tree)
+        assert isinstance(self.obj, Tree)
+        assert isinstance(other_obj, Tree)
         updated = update_meta(self.obj, other_obj)
         assert updated.hash_info == self.obj.hash_info
         self.obj = updated
@@ -1240,6 +1279,8 @@ META_SCHEMA = {
     Meta.PARAM_VERSION_ID: str,
 }
 
+CLOUD_SCHEMA = All({str: {**META_SCHEMA, **CHECKSUMS_SCHEMA}}, Length(max=1))
+
 ARTIFACT_SCHEMA = {
     **CHECKSUMS_SCHEMA,
     **META_SCHEMA,
@@ -1247,12 +1288,14 @@ ARTIFACT_SCHEMA = {
     Output.PARAM_PLOT: bool,
     Output.PARAM_PERSIST: bool,
     Output.PARAM_CHECKPOINT: bool,
+    Output.PARAM_CLOUD: CLOUD_SCHEMA,
 }
 
 DIR_FILES_SCHEMA: Dict[str, Any] = {
     **CHECKSUMS_SCHEMA,
     **META_SCHEMA,
     Required(Tree.PARAM_RELPATH): str,
+    Output.PARAM_CLOUD: CLOUD_SCHEMA,
 }
 
 SCHEMA = {

dvc/parsing/__init__.py

@@ -26,7 +26,6 @@ from .context import (
     KeyNotInContext,
     MergeError,
     Node,
-    SeqOrMap,
     VarsAlreadyLoaded,
 )
 from .interpolate import (
@@ -40,8 +39,10 @@ if TYPE_CHECKING:
     from typing import NoReturn
 
     from dvc.repo import Repo
+    from dvc.types import DictStrAny
+
+    from .context import SeqOrMap
 
-from dvc.types import DictStrAny
 
 logger = logging.getLogger(__name__)
 
@@ -92,7 +93,7 @@ def _reraise_err(
 
 
 def check_syntax_errors(
-    definition: DictStrAny, name: str, path: str, where: str = "stages"
+    definition: "DictStrAny", name: str, path: str, where: str = "stages"
 ):
     for key, d in definition.items():
         try:
@@ -111,8 +112,8 @@ def split_foreach_name(name: str) -> Tuple[str, Optional[str]]:
     return group, first(keys)
 
 
-def check_interpolations(data: DictStrAny, where: str, path: str):
-    def func(s: DictStrAny) -> None:
+def check_interpolations(data: "DictStrAny", where: str, path: str):
+    def func(s: "DictStrAny") -> None:
         if is_interpolated_string(s):
             raise ResolveError(
                 _format_preamble(f"'{where}'", path)
@@ -126,7 +127,7 @@ Definition = Union["ForeachDefinition", "EntryDefinition"]
 
 
 def make_definition(
-    resolver: "DataResolver", name: str, definition: DictStrAny, **kwargs
+    resolver: "DataResolver", name: str, definition: "DictStrAny", **kwargs
 ) -> Definition:
     args = resolver, resolver.context, name, definition
     if FOREACH_KWD in definition:
@@ -225,7 +226,7 @@ class EntryDefinition:
         resolver: DataResolver,
         context: Context,
         name: str,
-        definition: DictStrAny,
+        definition: "DictStrAny",
         where: str = STAGES_KWD,
     ):
         self.resolver = resolver
@@ -254,7 +255,7 @@ class EntryDefinition:
         except ContextError as exc:
             format_and_raise(exc, f"stage '{self.name}'", self.relpath)
 
-    def resolve_stage(self, skip_checks: bool = False) -> DictStrAny:
+    def resolve_stage(self, skip_checks: bool = False) -> "DictStrAny":
         context = self.context
         name = self.name
         if not skip_checks:
@@ -302,7 +303,7 @@ class EntryDefinition:
 
     def _resolve(
         self, context: "Context", value: Any, key: str, skip_checks: bool
-    ) -> DictStrAny:
+    ) -> "DictStrAny":
         try:
             return context.resolve(
                 value, skip_interpolation_checks=skip_checks, key=key
@@ -324,7 +325,7 @@ class ForeachDefinition:
         resolver: DataResolver,
         context: Context,
         name: str,
-        definition: DictStrAny,
+        definition: "DictStrAny",
         where: str = STAGES_KWD,
     ):
         self.resolver = resolver
@@ -349,7 +350,7 @@ class ForeachDefinition:
     def resolved_iterable(self):
         return self._resolve_foreach_data()
 
-    def _resolve_foreach_data(self) -> SeqOrMap:
+    def _resolve_foreach_data(self) -> "SeqOrMap":
         try:
             iterable = self.context.resolve(self.foreach_data, unwrap=False)
         except (ContextError, ParseError) as exc:
@@ -417,13 +418,13 @@ class ForeachDefinition:
     def _generate_name(self, key: str) -> str:
         return f"{self.name}{JOIN}{key}"
 
-    def resolve_all(self) -> DictStrAny:
+    def resolve_all(self) -> "DictStrAny":
         return join(map(self.resolve_one, self.normalized_iterable))
 
-    def resolve_one(self, key: str) -> DictStrAny:
+    def resolve_one(self, key: str) -> "DictStrAny":
         return self._each_iter(key)
 
-    def _each_iter(self, key: str) -> DictStrAny:
+    def _each_iter(self, key: str) -> "DictStrAny":
         err_message = f"Could not find '{key}' in foreach group '{self.name}'"
         with reraise(KeyError, EntryNotFound(err_message)):
             value = self.normalized_iterable[key]

dvc/parsing/context.py

@@ -228,7 +228,7 @@ class Container(Node, ABC):  # noqa: B024
 
         rem = rems[0]
         if not isinstance(d, Container):
-            raise ValueError(
+            raise ValueError(  # noqa: TRY004
                 f"{index} is a primitive value, cannot get '{rem}'"
             )
         return d.select(rem)

dvc/proc/manager.py

@@ -92,7 +92,7 @@ class ProcessManager:
 
         def handle_closed_process():
             logging.warning(
-                f"Process {name} had already aborted unexpectedly."
+                "Process %s had already aborted unexpectedly.", name
             )
             process_info.returncode = -1
             self[name] = process_info

dvc/prompt.py

@@ -25,8 +25,7 @@ def ask(prompt: str, limited_to: Optional[Collection[str]] = None):
             return answer
 
         logger.info(
-            "Your response must be one of: {options}. "
-            "Please try again.".format(options=limited_to)
+            "Your response must be one of: %s. Please try again.", limited_to
         )
 
 

dvc/repo/__init__.py

@@ -28,7 +28,7 @@ if TYPE_CHECKING:
     from dvc.fs.dvc import DVCFileSystem
     from dvc.lock import LockBase
     from dvc.machine import MachineManager
-    from dvc.scm import Base, Git, NoSCM
+    from dvc.scm import Git, NoSCM
     from dvc.stage import Stage
     from dvc.types import DictStrAny
     from dvc_data.hashfile.state import StateBase
@@ -171,7 +171,7 @@ class Repo:
         config: Optional["DictStrAny"] = None,
         url: Optional[str] = None,
         repo_factory: Optional[Callable] = None,
-        scm: Optional["Base"] = None,
+        scm: Optional[Union["Git", "NoSCM"]] = None,
     ):
         from dvc.config import Config
         from dvc.data_cloud import DataCloud
@@ -212,7 +212,7 @@ class Repo:
         # used by DVCFileSystem to determine if it should traverse subrepos
         self.subrepos = subrepos
 
-        self.cloud = DataCloud(self)
+        self.cloud: "DataCloud" = DataCloud(self)
         self.stage: "StageLoad" = StageLoad(self)
 
         self.lock: "LockBase"
@@ -275,7 +275,7 @@ class Repo:
             new.check_graph()
 
     @staticmethod
-    def open(url, *args, **kwargs):
+    def open(url, *args, **kwargs):  # noqa: A003
         if url is None:
             url = os.getcwd()
 

dvc/repo/add.py

@@ -194,7 +194,7 @@ def add(  # noqa: C901
                 stage.transfer(source, to_remote=to_remote, odb=odb, **kwargs)
             else:
                 try:
-                    stage.save(merge_versioned=True)
+                    stage.save()
                     if not no_commit:
                         stage.commit()
                 except CacheLinkError:

dvc/repo/brancher.py

@@ -32,7 +32,16 @@ def brancher(  # noqa: E302
             - empty string it there is no branches to iterate over
             - "workspace" if there are uncommitted changes in the SCM repo
     """
-    if not any([revs, all_branches, all_tags, all_commits, all_experiments]):
+    if not any(
+        [
+            revs,
+            all_branches,
+            all_tags,
+            all_commits,
+            all_experiments,
+            commit_date,
+        ]
+    ):
         yield ""
         return
 

dvc/repo/data.py

@@ -1,6 +1,5 @@
 import os
 import posixpath
-from collections import defaultdict
 from itertools import chain
 from typing import (
     TYPE_CHECKING,
@@ -10,124 +9,99 @@ from typing import (
     Iterator,
     List,
     Optional,
-    Tuple,
     TypedDict,
     Union,
 )
 
-from dvc.fs.git import GitFileSystem
-from dvc.ui import ui
-
 if TYPE_CHECKING:
     from dvc.repo import Repo
     from dvc.scm import Git, NoSCM
-    from dvc_data.hashfile.db import HashFileDB
-    from dvc_data.hashfile.hash_info import HashInfo
-    from dvc_data.hashfile.obj import HashFile
+    from dvc_data.index import DataIndex
 
 
 def posixpath_to_os_path(path: str) -> str:
     return path.replace(posixpath.sep, os.path.sep)
 
 
-def _in_cache(obj: "HashInfo", cache: "HashFileDB") -> bool:
-    from dvc_objects.errors import ObjectFormatError
+def _adapt_typ(typ):
+    from dvc_data.index.diff import ADD, DELETE, MODIFY
 
-    assert obj.value
-    try:
-        cache.check(obj.value)
-        return True
-    except (FileNotFoundError, ObjectFormatError):
-        return False
+    if typ == MODIFY:
+        return "modified"
+
+    if typ == ADD:
+        return "added"
+
+    if typ == DELETE:
+        return "deleted"
+
+    return typ
 
 
-def _shallow_diff(
-    root: str,
-    old: Optional["HashInfo"],
-    new: Optional["HashInfo"],
-    cache: "HashFileDB",
-) -> Dict[str, List[str]]:
-    d = {}
-
-    old_root = new_root = root
-    if old and old.isdir:
-        old_root = os.path.sep.join([root, ""])
-    if new and new.isdir:
-        new_root = os.path.sep.join([root, ""])
-
-    if old and not _in_cache(old, cache):
-        d["not_in_cache"] = [old_root]
-
-    if not old and not new:
-        return d
-    if not new:
-        return {"deleted": [old_root], **d}
-    if not old:
-        return {"added": [new_root], **d}
-    if old != new:
-        return {"modified": [new_root], **d}
-    return {"unchanged": [new_root], **d}
-
-
-def _granular_diff(
-    root: str,
-    old_obj: Optional["HashFile"],
-    new_obj: Optional["HashFile"],
-    cache: "HashFileDB",
-) -> Dict[str, List[str]]:
-    from dvc_data.hashfile.diff import ROOT
-    from dvc_data.hashfile.diff import diff as odiff
-
-    def path_join(root: str, *paths: str, isdir: bool = False) -> str:
-        if not isdir and paths == ROOT:
-            return root
-        return os.path.sep.join([root, *paths])
-
-    diff_data = odiff(old_obj, new_obj, cache)
-
-    output: Dict[str, List[str]] = defaultdict(list)
-    for state in ("added", "deleted", "modified", "unchanged"):
-        items = getattr(diff_data, state)
-        for item in items:  # pylint: disable=not-an-iterable
-            entry = item.old if state == "deleted" else item.new
-            isdir = entry.oid.isdir if entry.oid else False
-
-            path = path_join(root, *entry.key, isdir=isdir)
-            output[state].append(path)
-            if not item.old.in_cache and state != "added":
-                output["not_in_cache"].append(path)
-    return dict(output)
-
-
-def _get_obj_items(root: str, obj: Optional["HashFile"]) -> List[str]:
-    if not obj:
-        return []
-
-    from dvc_data.hashfile.tree import Tree
-
-    sep = os.path.sep
-    if isinstance(obj, Tree):
-        return [sep.join([root, *key]) for key, _, _ in obj]
-    return [root]
+def _adapt_path(change):
+    isdir = False
+    if change.new and change.new.meta:
+        isdir = change.new.meta.isdir
+    elif change.old and change.old.meta:
+        isdir = change.old.meta.isdir
+    key = change.key
+    if isdir:
+        key = (*key, "")
+    return os.path.sep.join(key)
 
 
 def _diff(
-    root: str,
-    old_oid: Optional["HashInfo"],
-    old_obj: Optional["HashFile"],
-    new_oid: Optional["HashInfo"],
-    new_obj: Optional["HashFile"],
-    odb: "HashFileDB",
-    granular: bool = False,
+    old: "DataIndex",
+    new: "DataIndex",
+    *,
+    granular: Optional[bool] = False,
+    with_missing: Optional[bool] = False,
 ) -> Dict[str, List[str]]:
-    if not granular:
-        return _shallow_diff(root, old_oid, new_oid, odb)
-    if (old_oid and not old_obj) or (new_oid and not new_obj):
-        # we don't have enough information to give full details
-        unknown = _get_obj_items(root, new_obj)
-        shallow_diff = _shallow_diff(root, old_oid, new_oid, odb)
-        return {**shallow_diff, "unknown": unknown}
-    return _granular_diff(root, old_obj, new_obj, odb)
+    from dvc_data.index.diff import UNCHANGED, UNKNOWN, diff
+
+    ret: Dict[str, List[str]] = {}
+
+    def _add_change(typ, change):
+        typ = _adapt_typ(typ)
+        if typ not in ret:
+            ret[typ] = []
+
+        ret[typ].append(_adapt_path(change))
+
+    for change in diff(
+        old,
+        new,
+        with_unchanged=True,
+        shallow=not granular,
+        hash_only=True,
+        with_unknown=True,
+    ):
+        if (
+            change.typ == UNCHANGED
+            and change.old
+            and change.new
+            and not change.old.hash_info
+            and not change.new.hash_info
+        ):
+            # NOTE: emulating previous behaviour
+            continue
+
+        if change.typ == UNKNOWN and not change.new:
+            # NOTE: emulating previous behaviour
+            continue
+
+        if (
+            with_missing
+            and change.old
+            and change.old.hash_info
+            and not old.odb_map[change.key].exists(change.old.hash_info.value)
+        ):
+            # NOTE: emulating previous behaviour
+            _add_change("not_in_cache", change)
+
+        _add_change(change.typ, change)
+
+    return ret
 
 
 class GitInfo(TypedDict, total=False):
@@ -169,81 +143,32 @@ def _git_info(
 
 
 def _diff_index_to_wtree(repo: "Repo", **kwargs: Any) -> Dict[str, List[str]]:
-    from dvc_data.hashfile.build import build
+    from .index import build_data_index
 
-    unstaged_diff = defaultdict(list)
-    for out in repo.index.outs:
-        if not out.use_cache:
-            continue
+    workspace = build_data_index(
+        repo.index, repo.root_dir, repo.fs, compute_hash=True
+    )
 
-        try:
-            _, _, new = build(
-                out.odb,
-                out.fs_path,
-                out.fs,
-                out.fs.PARAM_CHECKSUM,
-                ignore=out.dvcignore,
-                dry_run=True,
-            )
-        except FileNotFoundError:
-            new = None
-
-        cache = repo.odb.repo
-        root = str(out)
-        old = out.get_obj()
-
-        with ui.status(f"Calculating diff for {root} between index/workspace"):
-            d = _diff(
-                root,
-                out.hash_info,
-                old,
-                new.hash_info if new else None,
-                new,
-                cache,
-                **kwargs,
-            )
-
-        for state, items in d.items():
-            if not items:
-                continue
-            unstaged_diff[state].extend(items)
-    return unstaged_diff
+    return _diff(
+        repo.index.data["repo"],
+        workspace,
+        with_missing=True,
+        **kwargs,
+    )
 
 
 def _diff_head_to_index(
     repo: "Repo", head: str = "HEAD", **kwargs: Any
 ) -> Dict[str, List[str]]:
-    # we need to store objects from index and the HEAD to diff later
-    objs: Dict[str, Dict[str, Tuple[Optional["HashFile"], "HashInfo"]]]
-    objs = defaultdict(dict)
+    index = repo.index.data["repo"]
 
-    staged_diff = defaultdict(list)
     for rev in repo.brancher(revs=[head]):
-        for out in repo.index.outs:
-            if not out.use_cache:
-                continue
+        if rev == "workspace":
+            continue
 
-            root = str(out)
-            if isinstance(out.fs, GitFileSystem):
-                root = posixpath_to_os_path(root)
-            typ = "index" if rev == "workspace" else head
-            objs[root][typ] = (out.get_obj(), out.hash_info)
+        head_index = repo.index.data["repo"]
 
-    cache = repo.odb.repo
-    for root, obj_d in objs.items():
-        old_obj, old_oid = obj_d.get(head, (None, None))
-        new_obj, new_oid = obj_d.get("index", (None, None))
-        with ui.status(f"Calculating diff for {root} between head/index"):
-            d = _diff(
-                root, old_oid, old_obj, new_oid, new_obj, cache, **kwargs
-            )
-
-        for state, items in d.items():
-            if not items:
-                continue
-            staged_diff[state].extend(items)
-
-    return staged_diff
+    return _diff(head_index, index, **kwargs)
 
 
 class Status(TypedDict):
@@ -289,8 +214,6 @@ def status(repo: "Repo", untracked_files: str = "no", **kwargs: Any) -> Status:
     except (SCMError, NoSCMError):
         committed_diff = {}
     else:
-        # we don't care about not-in-cache between the head and the index.
-        committed_diff.pop("not_in_cache", None)
         unchanged &= set(committed_diff.pop("unchanged", []))
 
     git_info = _git_info(repo.scm, untracked_files=untracked_files)

dvc/repo/diff.py

@@ -229,7 +229,7 @@ def _filter_missing(dvcfs: "DVCFileSystem", paths: Set[str]) -> Iterator[str]:
             if (
                 entry
                 and info["type"] == "directory"
-                and not entry.odb.exists(entry.hash_info.value)
+                and not dvcfs.repo.odb.local.exists(entry.hash_info.value)
             ):
                 yield path
         except FileNotFoundError:

dvc/repo/experiments/__init__.py

@@ -2,7 +2,7 @@ import logging
 import os
 import re
 import time
-from typing import Dict, Iterable, Optional
+from typing import TYPE_CHECKING, Dict, Iterable, Optional
 
 from funcy import chain, first
 
@@ -18,7 +18,6 @@ from .exceptions import (
     MultipleBranchError,
 )
 from .executor.base import BaseExecutor
-from .queue.base import BaseStashQueue, QueueEntry
 from .queue.celery import LocalCeleryQueue
 from .queue.tempdir import TempDirQueue
 from .queue.workspace import WorkspaceQueue
@@ -32,9 +31,12 @@ from .refs import (
     WORKSPACE_STASH,
     ExpRefInfo,
 )
-from .stash import ExpStashEntry
 from .utils import check_ref_format, exp_refs_by_rev, unlocked_repo
 
+if TYPE_CHECKING:
+    from .queue.base import BaseStashQueue, QueueEntry
+    from .stash import ExpStashEntry
+
 logger = logging.getLogger(__name__)
 
 
@@ -85,7 +87,7 @@ class Experiments:
         return LocalCeleryQueue(self.repo, CELERY_STASH, CELERY_FAILED_STASH)
 
     @property
-    def stash_revs(self) -> Dict[str, ExpStashEntry]:
+    def stash_revs(self) -> Dict[str, "ExpStashEntry"]:
         revs = {}
         for queue in (self.workspace_queue, self.celery_queue):
             revs.update(queue.stash.stash_revs)
@@ -107,7 +109,7 @@ class Experiments:
                 )
                 self.scm.reset()
 
-        exp_queue: BaseStashQueue = (
+        exp_queue: "BaseStashQueue" = (
             self.tempdir_queue if tmp_dir else self.workspace_queue
         )
         self.queue_one(exp_queue, **kwargs)
@@ -119,11 +121,11 @@ class Experiments:
 
     def queue_one(
         self,
-        queue: BaseStashQueue,
+        queue: "BaseStashQueue",
         checkpoint_resume: Optional[str] = None,
         reset: bool = False,
         **kwargs,
-    ) -> QueueEntry:
+    ) -> "QueueEntry":
         """Queue a single experiment."""
         if reset:
             self.reset_checkpoints()
@@ -162,7 +164,7 @@ class Experiments:
         return None
 
     def reproduce_celery(  # noqa: C901
-        self, entries: Optional[Iterable[QueueEntry]] = None, **kwargs
+        self, entries: Optional[Iterable["QueueEntry"]] = None, **kwargs
     ) -> Dict[str, str]:
         results: Dict[str, str] = {}
         if entries is None:
@@ -238,11 +240,11 @@ class Experiments:
 
     def new(
         self,
-        queue: BaseStashQueue,
+        queue: "BaseStashQueue",
         *args,
         checkpoint_resume: Optional[str] = None,
         **kwargs,
-    ) -> QueueEntry:
+    ) -> "QueueEntry":
         """Create and enqueue a new experiment.
 
         Experiment will be derived from the current workspace.
@@ -266,11 +268,11 @@ class Experiments:
 
     def _resume_checkpoint(
         self,
-        queue: BaseStashQueue,
+        queue: "BaseStashQueue",
         *args,
         resume_rev: Optional[str] = None,
         **kwargs,
-    ) -> QueueEntry:
+    ) -> "QueueEntry":
         """Create and queue a resumed checkpoint experiment."""
         assert resume_rev
 
@@ -340,7 +342,7 @@ class Experiments:
 
     @unlocked_repo
     def _reproduce_queue(
-        self, queue: BaseStashQueue, **kwargs
+        self, queue: "BaseStashQueue", **kwargs
     ) -> Dict[str, str]:
         """Reproduce queued experiments.
 
@@ -424,7 +426,7 @@ class Experiments:
                 try:
                     name = ExpRefInfo.from_ref(ref).name
                 except InvalidExpRefError:
-                    pass
+                    pass  # noqa: S110
             if not name:
                 if rev in self.stash_revs:
                     name = self.stash_revs[rev].name

dvc/repo/experiments/apply.py

@@ -5,6 +5,7 @@ from typing import TYPE_CHECKING, Optional
 
 from dvc.repo import locked
 from dvc.repo.scm_context import scm_context
+from dvc.scm import Git
 from dvc.utils.fs import remove
 
 from .exceptions import (
@@ -16,8 +17,6 @@ from .executor.base import BaseExecutor
 from .refs import EXEC_APPLY
 
 if TYPE_CHECKING:
-    from scmrepo.git import Git
-
     from dvc.repo import Repo
     from dvc.repo.experiments import Experiments
 
@@ -36,6 +35,7 @@ def apply(repo: "Repo", rev: str, force: bool = True, **kwargs):  # noqa: C901
 
     is_stash: bool = False
 
+    assert isinstance(repo.scm, Git)
     try:
         exp_rev = resolve_rev(repo.scm, rev)
         exps.check_baseline(exp_rev)
@@ -90,6 +90,7 @@ def _apply_workspace(repo: "Repo", rev: str, force: bool):
     from scmrepo.exceptions import MergeConflictError
     from scmrepo.exceptions import SCMError as _SCMError
 
+    assert isinstance(repo.scm, Git)
     if repo.scm.is_dirty(untracked_files=True):
         logger.debug("Stashing workspace")
         stash_rev: Optional[str] = repo.scm.stash.push(include_untracked=True)

dvc/repo/experiments/exceptions.py

@@ -128,5 +128,9 @@ class UnresolvedQueueExpNamesError(UnresolvedExpNamesError):
     NAME = "queued experiment name"
 
 
+class UnresolvedRunningExpNamesError(UnresolvedExpNamesError):
+    NAME = "running experiment name"
+
+
 class ExpQueueEmptyError(DvcException):
     pass

dvc/repo/experiments/executor/base.py

@@ -43,10 +43,9 @@ from dvc.utils.fs import remove
 if TYPE_CHECKING:
     from queue import Queue
 
-    from scmrepo.git import Git
-
     from dvc.repo import Repo
     from dvc.repo.experiments.stash import ExpStashEntry
+    from dvc.scm import Git
     from dvc.stage import PipelineStage
 
 logger = logging.getLogger(__name__)

dvc/repo/experiments/executor/local.py

@@ -18,19 +18,18 @@ from dvc.repo.experiments.refs import (
     EXEC_MERGE,
     EXEC_NAMESPACE,
     EXPS_TEMP,
-    ExpRefInfo,
 )
 from dvc.repo.experiments.utils import EXEC_TMP_DIR, get_exp_rwlock
-from dvc.scm import SCM, GitMergeError
+from dvc.scm import SCM, Git, GitMergeError
 from dvc.utils.fs import remove
 from dvc.utils.objects import cached_property
 
 from .base import BaseExecutor, TaskStatus
 
 if TYPE_CHECKING:
-    from scmrepo.git import Git
 
     from dvc.repo import Repo
+    from dvc.repo.experiments.refs import ExpRefInfo
     from dvc.repo.experiments.stash import ExpStashEntry
     from dvc.scm import NoSCM
 
@@ -131,6 +130,7 @@ class TempDirExecutor(BaseLocalExecutor):
 
         # checkout EXEC_HEAD and apply EXEC_MERGE on top of it without
         # committing
+        assert isinstance(self.scm, Git)
         head = EXEC_BRANCH if branch else EXEC_HEAD
         self.scm.checkout(head, detach=True)
         merge_rev = self.scm.get_ref(EXEC_MERGE)
@@ -150,7 +150,9 @@ class TempDirExecutor(BaseLocalExecutor):
         with open(local_config, "w", encoding="utf-8") as fobj:
             fobj.write(f"[cache]\n    dir = {cache_dir}")
 
-    def init_cache(self, repo: "Repo", rev: str, run_cache: bool = True):
+    def init_cache(
+        self, repo: "Repo", rev: str, run_cache: bool = True  # noqa: ARG002
+    ):
         """Initialize DVC cache."""
         self._config(repo.odb.repo.path)
 
@@ -214,6 +216,8 @@ class WorkspaceExecutor(BaseLocalExecutor):
         if infofile:
             self.info.dump_json(infofile)
 
+        assert isinstance(self.scm, Git)
+
         with get_exp_rwlock(repo, writes=[EXEC_NAMESPACE]):
             scm.set_ref(EXEC_HEAD, entry.head_rev)
             scm.set_ref(EXEC_MERGE, stash_rev)

dvc/repo/experiments/executor/ssh.py

@@ -22,11 +22,10 @@ from .base import BaseExecutor, ExecutorInfo, ExecutorResult, TaskStatus
 if TYPE_CHECKING:
     from queue import Queue
 
-    from scmrepo.git import Git
-
     from dvc.repo import Repo
     from dvc.repo.experiments.refs import ExpRefInfo
     from dvc.repo.experiments.stash import ExpStashEntry
+    from dvc.scm import Git
 
 logger = logging.getLogger(__name__)
 
@@ -121,7 +120,7 @@ class SSHExecutor(BaseExecutor):
 
     def init_git(
         self,
-        repo: "Repo",
+        repo: "Repo",  # noqa: ARG002
         scm: "Git",
         stash_rev: str,
         entry: "ExpStashEntry",

dvc/repo/experiments/init.py

@@ -19,7 +19,6 @@ from funcy import compact, lremove, lsplit
 
 from dvc.exceptions import DvcException
 from dvc.stage import PipelineStage
-from dvc.types import OptStr
 
 if TYPE_CHECKING:
     from dvc.dvcfile import ProjectFile, SingleStageFile
@@ -47,7 +46,7 @@ def _prompts(
     ] = None,
     allow_omission: bool = True,
     stream: Optional[TextIO] = None,
-) -> Dict[str, OptStr]:
+) -> Dict[str, Optional[str]]:
     from dvc.ui.prompt import Prompt
 
     defaults = defaults or {}

dvc/repo/experiments/ls.py

@@ -1,10 +1,10 @@
 import logging
 from collections import defaultdict
+from typing import Optional
 
 from dvc.repo import locked
 from dvc.repo.scm_context import scm_context
 from dvc.scm import iter_revs
-from dvc.types import Optional
 
 from .utils import exp_refs_by_baseline
 

dvc/repo/experiments/queue/base.py

@@ -27,7 +27,7 @@ from dvc.repo.experiments.exceptions import (
     CheckpointExistsError,
     ExperimentExistsError,
 )
-from dvc.repo.experiments.executor.base import BaseExecutor, ExecutorResult
+from dvc.repo.experiments.executor.base import BaseExecutor
 from dvc.repo.experiments.executor.local import WorkspaceExecutor
 from dvc.repo.experiments.refs import ExpRefInfo
 from dvc.repo.experiments.stash import ExpStash, ExpStashEntry
@@ -44,10 +44,10 @@ from dvc.utils.objects import cached_property
 from .utils import get_remote_executor_refs
 
 if TYPE_CHECKING:
-    from scmrepo.git import Git
-
     from dvc.repo import Repo
     from dvc.repo.experiments import Experiments
+    from dvc.repo.experiments.executor.base import ExecutorResult
+    from dvc.scm import Git
 
 logger = logging.getLogger(__name__)
 
@@ -87,7 +87,7 @@ class QueueGetResult(NamedTuple):
 
 class QueueDoneResult(NamedTuple):
     entry: QueueEntry
-    result: Optional[ExecutorResult]
+    result: Optional["ExecutorResult"]
 
 
 class ExpRefAndQueueEntry(NamedTuple):
@@ -118,6 +118,9 @@ class BaseStashQueue(ABC):
 
     @property
     def scm(self) -> "Git":
+        from dvc.scm import Git
+
+        assert isinstance(self.repo.scm, Git)
         return self.repo.scm
 
     @cached_property
@@ -200,7 +203,7 @@ class BaseStashQueue(ABC):
 
         def _format_entry(
             entry: QueueEntry,
-            exp_result: Optional[ExecutorResult] = None,
+            exp_result: Optional["ExecutorResult"] = None,
             status: str = "Unknown",
         ) -> Dict[str, Any]:
             name = entry.name
@@ -258,7 +261,7 @@ class BaseStashQueue(ABC):
         """Reproduce queued experiments sequentially."""
 
     @abstractmethod
-    def get_result(self, entry: QueueEntry) -> Optional[ExecutorResult]:
+    def get_result(self, entry: QueueEntry) -> Optional["ExecutorResult"]:
         """Return result of the specified item.
 
         This method blocks until the specified item has been collected.
@@ -432,6 +435,7 @@ class BaseStashQueue(ABC):
                         name=name,
                     )
                     stash_rev = self.stash.push(message=msg)
+                    assert stash_rev
                     logger.debug(
                         (
                             "Stashed experiment '%s' with baseline '%s' "
@@ -632,7 +636,7 @@ class BaseStashQueue(ABC):
     def collect_git(
         exp: "Experiments",
         executor: BaseExecutor,
-        exec_result: ExecutorResult,
+        exec_result: "ExecutorResult",
     ) -> Dict[str, str]:
         results = {}
 
@@ -664,7 +668,7 @@ class BaseStashQueue(ABC):
         cls,
         exp: "Experiments",
         executor: BaseExecutor,
-        exec_result: ExecutorResult,
+        exec_result: "ExecutorResult",
     ) -> Dict[str, str]:
         results = cls.collect_git(exp, executor, exec_result)
 

dvc/repo/experiments/queue/celery.py

@@ -17,12 +17,14 @@ from typing import (
 
 from celery.result import AsyncResult
 from funcy import first
-from kombu.message import Message
 
 from dvc.daemon import daemonize
 from dvc.exceptions import DvcException
-from dvc.repo.experiments.exceptions import UnresolvedQueueExpNamesError
-from dvc.repo.experiments.executor.base import ExecutorInfo, ExecutorResult
+from dvc.repo.experiments.exceptions import (
+    UnresolvedQueueExpNamesError,
+    UnresolvedRunningExpNamesError,
+)
+from dvc.repo.experiments.executor.base import ExecutorInfo
 from dvc.repo.experiments.refs import CELERY_STASH
 from dvc.repo.experiments.utils import EXEC_TMP_DIR, get_exp_rwlock
 from dvc.ui import ui
@@ -33,23 +35,27 @@ from .base import (
     ExpRefAndQueueEntry,
     QueueDoneResult,
     QueueEntry,
-    QueueGetResult,
 )
 from .exceptions import CannotKillTasksError
 from .tasks import run_exp
 from .utils import fetch_running_exp_from_temp_dir
 
 if TYPE_CHECKING:
+    from kombu.message import Message
+
+    from dvc.repo.experiments.executor.base import ExecutorResult
     from dvc.repo.experiments.refs import ExpRefInfo
     from dvc_task.app import FSApp
     from dvc_task.proc.manager import ProcessManager
     from dvc_task.worker import TemporaryWorker
 
+    from .base import QueueGetResult
+
 logger = logging.getLogger(__name__)
 
 
 class _MessageEntry(NamedTuple):
-    msg: Message
+    msg: "Message"
     entry: QueueEntry
 
 
@@ -139,11 +145,11 @@ class LocalCeleryQueue(BaseStashQueue):
         cmd = ["exp", "queue-worker", node_name]
         name = f"dvc-exp-worker-{num}"
 
-        logger.debug(f"start a new worker: {name}, node: {node_name}")
+        logger.debug("start a new worker: %s, node: %s", name, node_name)
         if os.name == "nt":
             daemonize(cmd)
         else:
-            ManagedProcess.spawn(["dvc"] + cmd, wdir=self.wdir, name=name)
+            ManagedProcess.spawn(["dvc", *cmd], wdir=self.wdir, name=name)
 
     def start_workers(self, count: int) -> int:
         """start some workers to process the queued tasks.
@@ -155,7 +161,7 @@ class LocalCeleryQueue(BaseStashQueue):
             newly spawned worker number.
         """
 
-        logger.debug(f"Spawning {count} exp queue workers")
+        logger.debug("Spawning %s exp queue workers", count)
         active_worker: Dict = self.worker_status()
 
         started = 0
@@ -165,7 +171,7 @@ class LocalCeleryQueue(BaseStashQueue):
             ]
             node_name = f"dvc-exp-{wdir_hash}-{num}@localhost"
             if node_name in active_worker:
-                logger.debug(f"Exp queue worker {node_name} already exist")
+                logger.debug("Exp queue worker %s already exist", node_name)
                 continue
             self._spawn_worker(num)
             started += 1
@@ -181,7 +187,7 @@ class LocalCeleryQueue(BaseStashQueue):
 
     # NOTE: Queue consumption should not be done directly. Celery worker(s)
     # will automatically consume available experiments.
-    def get(self) -> QueueGetResult:
+    def get(self) -> "QueueGetResult":
         raise NotImplementedError
 
     def iter_queued(self) -> Generator[QueueEntry, None, None]:
@@ -317,7 +323,7 @@ class LocalCeleryQueue(BaseStashQueue):
                     self.proc.kill(queue_entry.stash_rev)
                 else:
                     self.proc.interrupt(queue_entry.stash_rev)
-                logger.debug(f"Task {rev} had been killed.")
+                ui.write(f"{rev} has been killed.")
             except ProcessLookupError:
                 fail_to_kill_entries[queue_entry] = rev
         return fail_to_kill_entries
@@ -327,7 +333,7 @@ class LocalCeleryQueue(BaseStashQueue):
     ) -> None:
         remained_revs: List[str] = []
         running_ids = self._get_running_task_ids()
-        logger.debug(f"Current running tasks ids: {running_ids}.")
+        logger.debug("Current running tasks ids: %s.", running_ids)
         for msg, entry in self._iter_processed():
             if entry not in remained_entries:
                 continue
@@ -338,8 +344,9 @@ class LocalCeleryQueue(BaseStashQueue):
                 result: AsyncResult = AsyncResult(task_id)
                 if not result.ready():
                     logger.debug(
-                        f"Task id {task_id} rev {remained_entries[entry]} "
-                        "marked as failure."
+                        "Task id %s rev %s marked as failure.",
+                        task_id,
+                        remained_entries[entry],
                     )
                     self.celery.backend.mark_as_failure(task_id, None)
 
@@ -373,12 +380,12 @@ class LocalCeleryQueue(BaseStashQueue):
             else:
                 to_kill[queue_entry] = rev
 
-        if missing_revs:
-            raise UnresolvedQueueExpNamesError(missing_revs)
-
         if to_kill:
             self._kill_entries(to_kill, force)
 
+        if missing_revs:
+            raise UnresolvedRunningExpNamesError(missing_revs)
+
     def shutdown(self, kill: bool = False):
         self.celery.control.shutdown()
         if kill:
@@ -406,8 +413,8 @@ class LocalCeleryQueue(BaseStashQueue):
             {rev}, self.iter_active(), self.iter_done()
         ).get(rev)
         if queue_entry is None:
-            if rev in self.match_queue_entry_by_name(
-                {rev}, self.iter_queued()
+            if self.match_queue_entry_by_name({rev}, self.iter_queued()).get(
+                rev
             ):
                 raise DvcException(
                     f"Experiment '{rev}' is in queue but has not been started"
@@ -438,8 +445,7 @@ class LocalCeleryQueue(BaseStashQueue):
     def worker_status(self) -> Dict[str, List[Dict]]:
         """Return the current active celery worker"""
         status = self.celery.control.inspect().active() or {}
-        logger.debug(f"Worker status: {status}")
-
+        logger.debug("Worker status: %s", status)
         return status
 
     def clear(self, *args, **kwargs):

dvc/repo/experiments/queue/tasks.py

@@ -74,7 +74,7 @@ def collect_exp(
         else:
             logger.debug("Experiment failed (Exec result was None)")
             celery_queue.stash_failed(entry)
-    except Exception:  # noqa: BLE001, pylint: disable=broad-except
+    except Exception:  # pylint: disable=broad-except
         # Log exceptions but do not re-raise so that task chain execution
         # continues
         logger.exception("Failed to collect experiment")

dvc/repo/experiments/queue/tempdir.py

@@ -7,12 +7,7 @@ from funcy import first
 
 from dvc.exceptions import DvcException
 from dvc.repo.experiments.exceptions import ExpQueueEmptyError
-from dvc.repo.experiments.executor.base import (
-    BaseExecutor,
-    ExecutorInfo,
-    ExecutorResult,
-    TaskStatus,
-)
+from dvc.repo.experiments.executor.base import ExecutorInfo, TaskStatus
 from dvc.repo.experiments.executor.local import TempDirExecutor
 from dvc.repo.experiments.utils import EXEC_PID_DIR, EXEC_TMP_DIR
 from dvc.utils.objects import cached_property
@@ -23,6 +18,7 @@ from .workspace import WorkspaceQueue
 
 if TYPE_CHECKING:
     from dvc.repo.experiments import Experiments
+    from dvc.repo.experiments.executor.base import BaseExecutor, ExecutorResult
     from dvc_task.proc.manager import ProcessManager
 
 logger = logging.getLogger(__name__)
@@ -96,7 +92,7 @@ class TempDirQueue(WorkspaceQueue):
                 )
 
     def _reproduce_entry(
-        self, entry: QueueEntry, executor: BaseExecutor
+        self, entry: QueueEntry, executor: "BaseExecutor"
     ) -> Dict[str, Dict[str, str]]:
         from dvc.stage.monitor import CheckpointKilledError
 
@@ -143,8 +139,8 @@ class TempDirQueue(WorkspaceQueue):
     @staticmethod
     def collect_executor(
         exp: "Experiments",
-        executor: BaseExecutor,
-        exec_result: ExecutorResult,
+        executor: "BaseExecutor",
+        exec_result: "ExecutorResult",
     ) -> Dict[str, str]:
         return BaseStashQueue.collect_executor(exp, executor, exec_result)
 

dvc/repo/experiments/queue/utils.py

@@ -16,7 +16,7 @@ logger = logging.getLogger(__name__)
 
 
 if TYPE_CHECKING:
-    from scmrepo.git import Git
+    from dvc.scm import Git
 
     from .base import BaseStashQueue
 

dvc/repo/experiments/queue/workspace.py

@@ -10,12 +10,7 @@ from voluptuous import Invalid
 from dvc.exceptions import DvcException
 from dvc.lock import make_lock
 from dvc.repo.experiments.exceptions import ExpQueueEmptyError
-from dvc.repo.experiments.executor.base import (
-    BaseExecutor,
-    ExecutorInfo,
-    ExecutorResult,
-    TaskStatus,
-)
+from dvc.repo.experiments.executor.base import ExecutorInfo, TaskStatus
 from dvc.repo.experiments.executor.local import WorkspaceExecutor
 from dvc.repo.experiments.refs import EXEC_BRANCH, WORKSPACE_STASH
 from dvc.repo.experiments.utils import get_exp_rwlock, to_studio_params
@@ -23,10 +18,13 @@ from dvc.rwlock import RWLOCK_FILE, RWLOCK_LOCK, SCHEMA
 from dvc.utils import relpath
 from dvc.utils.fs import remove
 
-from .base import BaseStashQueue, QueueDoneResult, QueueEntry, QueueGetResult
+from .base import BaseStashQueue, QueueEntry, QueueGetResult
 
 if TYPE_CHECKING:
     from dvc.repo.experiments import Experiments
+    from dvc.repo.experiments.executor.base import BaseExecutor, ExecutorResult
+
+    from .base import QueueDoneResult
 
 logger = logging.getLogger(__name__)
 
@@ -74,13 +72,13 @@ class WorkspaceQueue(BaseStashQueue):
         # need to be handled via the queue
         raise NotImplementedError
 
-    def iter_done(self) -> Generator[QueueDoneResult, None, None]:
+    def iter_done(self) -> Generator["QueueDoneResult", None, None]:
         raise NotImplementedError
 
-    def iter_failed(self) -> Generator[QueueDoneResult, None, None]:
+    def iter_failed(self) -> Generator["QueueDoneResult", None, None]:
         raise NotImplementedError
 
-    def iter_success(self) -> Generator[QueueDoneResult, None, None]:
+    def iter_success(self) -> Generator["QueueDoneResult", None, None]:
         raise NotImplementedError
 
     def reproduce(self) -> Dict[str, Dict[str, str]]:
@@ -94,7 +92,7 @@ class WorkspaceQueue(BaseStashQueue):
         return results
 
     def _reproduce_entry(
-        self, entry: QueueEntry, executor: BaseExecutor
+        self, entry: QueueEntry, executor: "BaseExecutor"
     ) -> Dict[str, Dict[str, str]]:
         from dvc_studio_client.post_live_metrics import post_live_metrics
 
@@ -152,8 +150,8 @@ class WorkspaceQueue(BaseStashQueue):
     @staticmethod
     def collect_executor(  # pylint: disable=unused-argument
         exp: "Experiments",
-        executor: BaseExecutor,  # noqa: ARG004
-        exec_result: ExecutorResult,
+        executor: "BaseExecutor",  # noqa: ARG004
+        exec_result: "ExecutorResult",
     ) -> Dict[str, str]:
         results: Dict[str, str] = {}
         exp_rev = exp.scm.get_ref(EXEC_BRANCH)
@@ -164,7 +162,7 @@ class WorkspaceQueue(BaseStashQueue):
 
         return results
 
-    def get_result(self, entry: QueueEntry) -> Optional[ExecutorResult]:
+    def get_result(self, entry: QueueEntry) -> Optional["ExecutorResult"]:
         raise NotImplementedError
 
     def kill(self, revs: Collection[str]) -> None:
@@ -239,7 +237,10 @@ class WorkspaceQueue(BaseStashQueue):
                 remove(path)
             return False
 
-    def get_running_exps(self, fetch_refs: bool = True) -> Dict[str, Dict]:
+    def get_running_exps(
+        self,
+        fetch_refs: bool = True,  # noqa: ARG002
+    ) -> Dict[str, Dict]:
         from dvc.utils.serialize import load_json
 
         assert self._EXEC_NAME

dvc/repo/experiments/remove.py

@@ -3,19 +3,17 @@ from typing import TYPE_CHECKING, Dict, Iterable, List, Optional, Union
 
 from dvc.repo import locked
 from dvc.repo.scm_context import scm_context
-from dvc.scm import iter_revs
+from dvc.scm import Git, iter_revs
 
 from .exceptions import UnresolvedExpNamesError
-from .queue.base import QueueEntry
-from .refs import ExpRefInfo
 from .utils import exp_refs, exp_refs_by_baseline, push_refspec
 
 if TYPE_CHECKING:
     from dvc.repo import Repo
     from dvc.repo.experiments.queue.celery import LocalCeleryQueue
-    from dvc.scm import Git
 
-    from .queue.base import ExpRefAndQueueEntry
+    from .queue.base import ExpRefAndQueueEntry, QueueEntry
+    from .refs import ExpRefInfo
 
 
 logger = logging.getLogger(__name__)
@@ -39,6 +37,8 @@ def remove(  # noqa: C901
 
     if queue:
         removed.extend(celery_queue.clear(queued=True))
+
+    assert isinstance(repo.scm, Git)
     if all_commits:
         removed.extend(
             _remove_commited_exps(
@@ -99,7 +99,7 @@ def _resolve_exp_by_baseline(
 
 
 def _remove_commited_exps(
-    scm: "Git", exp_refs_list: Iterable[ExpRefInfo], remote: Optional[str]
+    scm: "Git", exp_refs_list: Iterable["ExpRefInfo"], remote: Optional[str]
 ) -> List[str]:
     if remote:
         from dvc.scm import TqdmGit

dvc/repo/experiments/save.py

@@ -2,6 +2,8 @@ import logging
 import os
 from typing import TYPE_CHECKING, List, Optional
 
+from dvc.scm import Git
+
 from .exceptions import ExperimentExistsError
 from .refs import ExpRefInfo
 from .utils import check_ref_format, get_random_exp_name
@@ -29,6 +31,8 @@ def _save_experiment(
     if repo.scm.get_ref(ref) and not force:
         raise ExperimentExistsError(ref_info.name, command="save")
 
+    assert isinstance(repo.scm, Git)
+
     repo.scm.add([], update=True)
     if include_untracked:
         repo.scm.add(include_untracked)
@@ -51,6 +55,8 @@ def save(
     """
     logger.debug("Saving workspace in %s", os.getcwd())
 
+    assert isinstance(repo.scm, Git)
+
     _, _, untracked = repo.scm.status()
     if include_untracked:
         untracked = [

dvc/repo/experiments/show.py

@@ -20,13 +20,12 @@ from scmrepo.exceptions import SCMError as InnerScmError
 
 from dvc.repo.metrics.show import _gather_metrics
 from dvc.repo.params.show import _gather_params
-from dvc.scm import SCMError, iter_revs, resolve_rev
+from dvc.scm import Git, SCMError, iter_revs, resolve_rev
 from dvc.utils import error_handler, onerror_collect, relpath
 
 from .refs import ExpRefInfo
 
 if TYPE_CHECKING:
-    from scmrepo.git import Git
     from scmrepo.git.objects import GitCommit
 
     from dvc.repo import Repo
@@ -190,6 +189,7 @@ def _collect_branch(
     ref_info = ExpRefInfo(baseline_sha=baseline)
     commits: List[Tuple[str, "GitCommit", str, List[str]]] = []
 
+    assert isinstance(repo.scm, Git)
     for ref in repo.scm.iter_refs(base=str(ref_info)):
         try:
             commit = repo.scm.resolve_commit(ref)
@@ -414,6 +414,8 @@ def show(
     if isinstance(revs, str):
         revs = [revs]
 
+    assert isinstance(repo.scm, Git)
+
     found_revs: Dict[str, List[str]] = {"workspace": []}
     found_revs.update(
         iter_revs(repo.scm, revs, num, all_branches, all_tags, all_commits)

dvc/repo/experiments/utils.py

@@ -17,11 +17,10 @@ from typing import (
     Union,
 )
 
-from scmrepo.git import Git
-
 from dvc.exceptions import InvalidArgumentError
 from dvc.repo.experiments.exceptions import AmbiguousExpRefInfo
 from dvc.rwlock import rwlock
+from dvc.scm import Git
 
 from .refs import (
     EXEC_APPLY,

dvc/repo/gc.py

@@ -72,7 +72,7 @@ def gc(
         for repo in all_repos:
             stack.enter_context(repo.lock)
 
-        for repo in all_repos + [self]:
+        for repo in [*all_repos, self]:
             for obj_ids in repo.used_objs(
                 all_branches=all_branches,
                 with_deps=with_deps,

dvc/repo/imp_url.py

@@ -33,9 +33,6 @@ def imp_url(  # noqa: C901
     fs_config=None,
     version_aware: bool = False,
 ):
-    from dvc.dvcfile import load_file
-    from dvc.stage import Stage, create_stage, restore_fields
-
     out = resolve_output(url, out)
     path, wdir, out = resolve_paths(
         self, out, always_local=to_remote and not out
@@ -64,22 +61,19 @@ def imp_url(  # noqa: C901
             fs_config = {}
         fs_config["version_aware"] = True
 
-    stage = create_stage(
-        Stage,
-        self,
-        fname or path,
+    stage = self.stage.create(
+        single_stage=True,
+        validate=False,
+        fname=fname or path,
         wdir=wdir,
         deps=[url],
         outs=[out],
         erepo=erepo,
         fs_config=fs_config,
     )
-    restore_fields(stage)
 
     out_obj = stage.outs[0]
     out_obj.annot.update(desc=desc, type=type, labels=labels, meta=meta)
-    dvcfile = load_file(self, stage.path)
-    dvcfile.remove()
 
     try:
         self.check_graph(stages={stage})
@@ -96,13 +90,10 @@ def imp_url(  # noqa: C901
         stage.save_deps()
         stage.md5 = stage.compute_md5()
     else:
+        if stage.deps[0].fs.version_aware:
+            stage.outs[0].can_push = False
         stage.run(jobs=jobs, no_download=no_download)
 
-    if not no_exec and stage.deps[0].fs.version_aware:
-        stage.outs[0].can_push = False
-
     stage.frozen = frozen
-
-    dvcfile.dump(stage)
-
+    stage.dump()
     return stage

dvc/repo/index.py

@@ -128,10 +128,14 @@ class Index:
         self._params = params or {}
         self._collected_targets: Dict[int, List["StageInfo"]] = {}
 
-    def __repr__(self) -> str:
-        rev = "workspace"
+    @cached_property
+    def rev(self) -> Optional[str]:
         if not isinstance(self.repo.fs, LocalFileSystem):
-            rev = self.repo.get_rev()[:7]
+            return self.repo.get_rev()[:7]
+        return None
+
+    def __repr__(self) -> str:
+        rev = self.rev or "workspace"
         return f"Index({self.repo}, fs@{rev})"
 
     @classmethod
@@ -259,10 +263,29 @@ class Index:
                     sources.append(plot_id)
         return sources
 
+    @cached_property
+    def data_keys(self) -> Dict[str, Set["DataIndexKey"]]:
+        from collections import defaultdict
+
+        by_workspace: Dict[str, Set["DataIndexKey"]] = defaultdict(set)
+
+        by_workspace["repo"] = set()
+        by_workspace["local"] = set()
+
+        for out in self.outs:
+            if not out.use_cache:
+                continue
+
+            workspace, key = out.index_key
+            by_workspace[workspace].add(key)
+
+        return dict(by_workspace)
+
     @cached_property
     def data(self) -> "Dict[str, DataIndex]":
         from collections import defaultdict
 
+        from dvc.config import NoRemoteError
         from dvc_data.index import DataIndex
 
         by_workspace: dict = defaultdict(DataIndex)
@@ -281,6 +304,14 @@ class Index:
             entry.key = key
             data_index.add(entry)
 
+            data_index.odb_map[key] = out.odb
+            try:
+                data_index.remote_map[key] = self.repo.cloud.get_remote_odb(
+                    out.remote
+                )
+            except NoRemoteError:
+                pass
+
         return dict(by_workspace)
 
     @staticmethod
@@ -297,24 +328,32 @@ class Index:
         )
 
     def collect_targets(
-        self, targets: Optional["TargetType"], **kwargs: Any
+        self, targets: Optional["TargetType"], *, onerror=None, **kwargs: Any
     ) -> List["StageInfo"]:
-        from itertools import chain
-
+        from dvc.exceptions import DvcException
         from dvc.repo.stage import StageInfo
         from dvc.utils.collections import ensure_list
 
+        if not onerror:
+
+            def onerror(_target, _exc):
+                raise  # pylint: disable=misplaced-bare-raise
+
         targets = ensure_list(targets)
         if not targets:
             return [StageInfo(stage) for stage in self.stages]
         targets_hash = self._hash_targets(targets, **kwargs)
         if targets_hash not in self._collected_targets:
-            self._collected_targets[targets_hash] = list(
-                chain.from_iterable(
-                    self.repo.stage.collect_granular(target, **kwargs)
-                    for target in targets
-                )
-            )
+            collected = []
+            for target in targets:
+                try:
+                    collected.extend(
+                        self.repo.stage.collect_granular(target, **kwargs)
+                    )
+                except DvcException as exc:
+                    onerror(target, exc)
+            self._collected_targets[targets_hash] = collected
+
         return self._collected_targets[targets_hash]
 
     def used_objs(
@@ -395,6 +434,10 @@ class IndexView:
         self.stages = list({stage for stage, _ in stage_infos})
         self._outs_filter = outs_filter
 
+    @property
+    def repo(self) -> "Repo":
+        return self._index.repo
+
     @property
     def deps(self) -> Iterator["Dependency"]:
         for stage in self.stages:
@@ -429,6 +472,20 @@ class IndexView:
             )
         return prefixes
 
+    @cached_property
+    def data_keys(self) -> Dict[str, Set["DataIndexKey"]]:
+        from collections import defaultdict
+
+        ret: Dict[str, Set["DataIndexKey"]] = defaultdict(set)
+
+        for out, filter_info in self._filtered_outs:
+            workspace, key = out.index_key
+            if filter_info and out.fs.path.isin(filter_info, out.fs_path):
+                key = key + out.fs.path.relparts(filter_info, out.fs_path)
+            ret[workspace].add(key)
+
+        return dict(ret)
+
     @cached_property
     def data(self) -> Dict[str, Union["DataIndex", "DataIndexView"]]:
         from dvc_data.index import DataIndex, view
@@ -458,44 +515,50 @@ def build_data_index(
     index: Union["Index", "IndexView"],
     path: str,
     fs: "FileSystem",
-    workspace: Optional[str] = "repo",
+    workspace: str = "repo",
+    compute_hash: Optional[bool] = False,
 ) -> "DataIndex":
     from dvc_data.index import DataIndex, DataIndexEntry
     from dvc_data.index.build import build_entries, build_entry
+    from dvc_data.index.save import build_tree
 
     data = DataIndex()
-    for out in index.outs:
-        if not out.use_cache:
-            continue
-
-        ws, key = out.index_key
-        if ws != workspace:
-            continue
-
-        parts = out.fs.path.relparts(out.fs_path, out.repo.root_dir)
-        out_path = fs.path.join(path, *parts)
+    for key in index.data_keys.get(workspace, set()):
+        out_path = fs.path.join(path, *key)
 
         try:
-            entry = build_entry(out_path, fs)
+            out_entry = build_entry(
+                out_path,
+                fs,
+                compute_hash=compute_hash,
+                state=index.repo.state,
+            )
         except FileNotFoundError:
-            entry = DataIndexEntry(path=out_path, fs=fs)
+            out_entry = DataIndexEntry(path=out_path, fs=fs)
 
-        entry.key = key
+        out_entry.key = key
 
-        if not entry.meta or not entry.meta.isdir:
-            data.add(entry)
+        if not out_entry.meta or not out_entry.meta.isdir:
+            data.add(out_entry)
             continue
 
-        entry.loaded = True
-        data.add(entry)
-
-        for entry in build_entries(out_path, fs):
+        for entry in build_entries(
+            out_path, fs, compute_hash=compute_hash, state=index.repo.state
+        ):
             if not entry.key or entry.key == ("",):
                 # NOTE: whether the root will be returned by build_entries
                 # depends on the filesystem (e.g. local doesn't, but s3 does).
-                entry.key = key
-            else:
-                entry.key = key + entry.key
+                continue
+
+            entry.key = key + entry.key
             data.add(entry)
 
+        if compute_hash:
+            tree_meta, tree = build_tree(data, key)
+            out_entry.meta = tree_meta
+            out_entry.hash_info = tree.hash_info
+
+        out_entry.loaded = True
+        data.add(out_entry)
+
     return data

dvc/repo/install.py

@@ -3,9 +3,8 @@ from typing import TYPE_CHECKING
 from dvc.exceptions import DvcException
 
 if TYPE_CHECKING:
-    from scmrepo.git import Git
-
     from dvc.repo import Repo
+    from dvc.scm import Git
 
 
 def pre_commit_install(scm: "Git") -> None:
@@ -68,7 +67,7 @@ def install(self: "Repo", use_pre_commit_tool: bool = False) -> None:
     If use_pre_commit_tool is set and pre-commit is installed it will be used
     to install the hooks.
     """
-    from scmrepo.git import Git
+    from dvc.scm import Git
 
     scm = self.scm
     if not isinstance(scm, Git):

dvc/repo/metrics/show.py

@@ -1,11 +1,10 @@
 import logging
 import os
-from typing import List
+from typing import TYPE_CHECKING, List
 
 from scmrepo.exceptions import SCMError
 
 from dvc.fs.dvc import DVCFileSystem
-from dvc.output import Output
 from dvc.repo import locked
 from dvc.repo.collect import collect
 from dvc.scm import NoSCMError
@@ -18,14 +17,17 @@ from dvc.utils import (
 from dvc.utils.collections import ensure_list
 from dvc.utils.serialize import load_path
 
+if TYPE_CHECKING:
+    from dvc.output import Output
+
 logger = logging.getLogger(__name__)
 
 
-def _is_metric(out: Output) -> bool:
+def _is_metric(out: "Output") -> bool:
     return bool(out.metric)
 
 
-def _to_fs_paths(metrics: List[Output]) -> List["str"]:
+def _to_fs_paths(metrics: List["Output"]) -> List["str"]:
     result = []
     for out in metrics:
         if out.metric:
@@ -146,7 +148,7 @@ def show(
         except (SCMError, NoSCMError):
             # SCMError - detached head
             # NoSCMError - no repo case
-            pass
+            pass  # noqa: S110
         else:
             if res.get("workspace") == res.get(active_branch):
                 res.pop("workspace", None)

dvc/repo/params/show.py

@@ -171,7 +171,7 @@ def show(
         except (SCMError, NoSCMError):
             # SCMError - detached head
             # NoSCMError - no repo case
-            pass
+            pass  # noqa: S110
         else:
             if res.get("workspace") == res.get(active_branch):
                 res.pop("workspace", None)

dvc/repo/plots/__init__.py

@@ -32,7 +32,7 @@ if TYPE_CHECKING:
     from dvc.fs import FileSystem
     from dvc.output import Output
     from dvc.repo import Repo
-    from dvc.types import DictAny, DictStrAny, StrPath
+    from dvc.types import DictStrAny, StrPath
 
 dpath.options.ALLOW_EMPTY_STRING_KEYS = True
 
@@ -424,7 +424,7 @@ def _adjust_sources(fs, plot_props, config_dir):
 def _resolve_definitions(
     fs: "FileSystem",
     targets: List[str],
-    props: "DictAny",
+    props: Dict[str, Any],
     config_path: "StrPath",
     definitions: "DictStrAny",
     onerror: Optional[Callable[[Any], Any]] = None,

dvc/repo/scm_context.py

@@ -18,9 +18,8 @@ from dvc.utils import relpath
 from dvc.utils.collections import ensure_list
 
 if TYPE_CHECKING:
-    from scmrepo.base import Base
-
     from dvc.repo import Repo
+    from dvc.scm import Base
 
 
 logger = logging.getLogger(__name__)
@@ -133,7 +132,7 @@ class SCMContext:
             and logger.isEnabledFor(logging.INFO)
         ):
             add_cmd = self._make_git_add_cmd(self.files_to_track)
-            logger.info(f"\nTo track the changes with git, run:\n\n{add_cmd}")
+            logger.info("\nTo track the changes with git, run:\n\n%s", add_cmd)
             logger.info(
                 "\nTo enable auto staging, run:\n\n"
                 "\tdvc config core.autostage true"

dvc/repo/stage.py

@@ -18,7 +18,6 @@ if typing.TYPE_CHECKING:
     from dvc.repo import Repo
     from dvc.stage import PipelineStage, Stage
     from dvc.stage.loader import StageLoader
-    from dvc.types import OptStr
 
 PROJECT_FILE = "dvc.yaml"
 
@@ -61,7 +60,7 @@ def _collect_specific_target(
     target: str,
     with_deps: bool,
     recursive: bool,
-) -> Tuple[StageIter, "OptStr", "OptStr"]:
+) -> Tuple[StageIter, Optional[str], Optional[str]]:
     from dvc.dvcfile import is_valid_filename
 
     # Optimization: do not collect the graph for a specific target
@@ -97,7 +96,10 @@ def locked(f):
 class StageLoad:
     def __init__(self, repo: "Repo") -> None:
         self.repo: "Repo" = repo
-        self.fs = repo.fs
+
+    @property
+    def fs(self):
+        return self.repo.fs
 
     @locked
     def add(

dvc/repo/worktree.py

@@ -1,6 +1,18 @@
 import logging
 from functools import partial
-from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Union, cast
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Dict,
+    Iterable,
+    Optional,
+    Set,
+    Tuple,
+    Union,
+    cast,
+)
+
+from funcy import first
 
 from dvc.fs.callbacks import Callback
 
@@ -28,11 +40,44 @@ def _meta_checksum(fs: "FileSystem", meta: "Meta") -> Any:
     return getattr(meta, cast(str, fs.PARAM_CHECKSUM))
 
 
+def worktree_view_by_remotes(
+    index: "Index",
+    targets: Optional["TargetType"] = None,
+    push: bool = False,
+    **kwargs: Any,
+) -> Iterable[Tuple[Optional[str], "IndexView"]]:
+    # pylint: disable=protected-access
+
+    from dvc.repo.index import IndexView
+
+    def outs_filter(view: "IndexView", remote: Optional[str]):
+        def _filter(out: "Output") -> bool:
+            if out.remote != remote:
+                return False
+            if view._outs_filter:
+                return view._outs_filter(out)
+            return True
+
+        return _filter
+
+    view = worktree_view(index, targets=targets, push=push, **kwargs)
+    remotes = {out.remote for out in view.outs}
+
+    if len(remotes) <= 1:
+        yield first(remotes), view
+        return
+
+    for remote in remotes:
+        yield remote, IndexView(
+            index, view._stage_infos, outs_filter(view, remote)
+        )
+
+
 def worktree_view(
     index: "Index",
     targets: Optional["TargetType"] = None,
     push: bool = False,
-    **kwargs,
+    **kwargs: Any,
 ) -> "IndexView":
     """Return view of data that can be stored in worktree remotes.
 
@@ -65,29 +110,44 @@ def worktree_view(
     )
 
 
+def _get_remote(
+    repo: "Repo", name: Optional[str], default: "Remote", command: str
+) -> "Remote":
+    if name in (None, default.name):
+        return default
+    return repo.cloud.get_remote(name, command)
+
+
 def fetch_worktree(
     repo: "Repo",
     remote: "Remote",
     targets: Optional["TargetType"] = None,
     jobs: Optional[int] = None,
-    **kwargs,
+    **kwargs: Any,
 ) -> int:
     from dvc_data.index import save
 
-    view = worktree_view(repo.index, targets=targets, **kwargs)
-    index = view.data["repo"]
-    for key, entry in index.iteritems():
-        entry.fs = remote.fs
-        entry.path = remote.fs.path.join(
-            remote.path,
-            *key,
-        )
-    total = len(index)
-    with Callback.as_tqdm_callback(
-        unit="file", desc="Fetch", disable=total == 0
-    ) as cb:
-        cb.set_size(total)
-        return save(index, callback=cb, jobs=jobs)
+    transferred = 0
+    for remote_name, view in worktree_view_by_remotes(
+        repo.index, push=True, targets=targets, **kwargs
+    ):
+        remote_obj = _get_remote(repo, remote_name, remote, "fetch")
+        index = view.data["repo"]
+        for key, entry in index.iteritems():
+            entry.fs = remote_obj.fs
+            entry.path = remote_obj.fs.path.join(
+                remote_obj.path,
+                *key,
+            )
+        total = len(index)
+        with Callback.as_tqdm_callback(
+            unit="file",
+            desc=f"Fetching from remote {remote_obj.name!r}",
+            disable=total == 0,
+        ) as cb:
+            cb.set_size(total)
+            transferred += save(index, callback=cb, jobs=jobs)
+    return transferred
 
 
 def push_worktree(
@@ -95,61 +155,70 @@ def push_worktree(
     remote: "Remote",
     targets: Optional["TargetType"] = None,
     jobs: Optional[int] = None,
-    **kwargs,
+    **kwargs: Any,
 ) -> int:
     from dvc.repo.index import build_data_index
     from dvc_data.index import checkout
 
-    view = worktree_view(
-        repo.index,
-        push=True,
-        targets=targets,
-        **kwargs,
-    )
-    new_index = view.data["repo"]
-    if remote.worktree:
-        logger.debug("indexing latest worktree for '%s'", remote.path)
-        old_index = build_data_index(view, remote.path, remote.fs)
-        logger.debug("Pushing worktree changes to '%s'", remote.path)
-    else:
-        old_index = None
-        logger.debug("Pushing version-aware files to '%s'", remote.path)
+    pushed = 0
+    stages: Set["Stage"] = set()
 
-    if remote.worktree:
-        diff_kwargs: Dict[str, Any] = {
-            "meta_only": True,
-            "meta_cmp_key": partial(_meta_checksum, remote.fs),
-        }
-    else:
-        diff_kwargs = {}
+    for remote_name, view in worktree_view_by_remotes(
+        repo.index, push=True, targets=targets, **kwargs
+    ):
+        remote_obj = _get_remote(repo, remote_name, remote, "push")
+        new_index = view.data["repo"]
+        if remote_obj.worktree:
+            logger.debug("indexing latest worktree for '%s'", remote_obj.path)
+            old_index = build_data_index(view, remote_obj.path, remote_obj.fs)
+            logger.debug("Pushing worktree changes to '%s'", remote_obj.path)
+        else:
+            old_index = None
+            logger.debug(
+                "Pushing version-aware files to '%s'", remote_obj.path
+            )
 
-    total = len(new_index)
-    with Callback.as_tqdm_callback(
-        unit="file", desc="Push", disable=total == 0
-    ) as cb:
-        cb.set_size(total)
-        pushed = checkout(
-            new_index,
-            remote.path,
-            remote.fs,
-            old=old_index,
-            delete=remote.worktree,
-            callback=cb,
-            latest_only=remote.worktree,
-            jobs=jobs,
-            **diff_kwargs,
-        )
-    if pushed:
-        for stage in view.stages:
-            for out in stage.outs:
-                workspace, _key = out.index_key
-                _update_out_meta(out, repo.index.data[workspace])
-            stage.dump(with_files=True, update_pipeline=False)
+        if remote_obj.worktree:
+            diff_kwargs: Dict[str, Any] = {
+                "meta_only": True,
+                "meta_cmp_key": partial(_meta_checksum, remote_obj.fs),
+            }
+        else:
+            diff_kwargs = {}
+
+        total = len(new_index)
+        with Callback.as_tqdm_callback(
+            unit="file",
+            desc=f"Pushing to remote {remote_obj.name!r}",
+            disable=total == 0,
+        ) as cb:
+            cb.set_size(total)
+            pushed += checkout(
+                new_index,
+                remote_obj.path,
+                remote_obj.fs,
+                old=old_index,
+                delete=remote_obj.worktree,
+                callback=cb,
+                latest_only=remote_obj.worktree,
+                jobs=jobs,
+                **diff_kwargs,
+            )
+
+        for out in view.outs:
+            workspace, _key = out.index_key
+            _update_out_meta(out, repo.index.data[workspace], remote_obj.name)
+            stages.add(out.stage)
+
+    for stage in stages:
+        stage.dump(with_files=True, update_pipeline=False)
     return pushed
 
 
 def _update_out_meta(
-    out: "Output", index: Union["DataIndex", "DataIndexView"]
+    out: "Output",
+    index: Union["DataIndex", "DataIndexView"],
+    remote: Optional[str] = None,
 ):
     from dvc_data.index.save import build_tree
 
@@ -168,6 +237,8 @@ def _update_out_meta(
                 repo.fs.path.relparts(fs_path, out.fs_path)
             ) or (None, None)
             entry.hash_info = hash_info
+            if entry.meta:
+                entry.meta.remote = remote
             if meta is not None and meta.version_id is not None:
                 # preserve existing version IDs for unchanged files in
                 # this dir (entry will have the latest remote version
@@ -182,6 +253,8 @@ def _update_out_meta(
             out.hash_info = entry.hash_info
         if out.meta.version_id is None:
             out.meta = entry.meta
+    if out.meta:
+        out.meta.remote = remote
 
 
 def update_worktree_stages(
@@ -228,7 +301,7 @@ def update_worktree_stages(
                 )
                 continue
             _fetch_out_changes(out, local_index, remote_index, remote)
-        stage.save(merge_versioned=True)
+        stage.save()
         for out in stage.outs:
             _update_out_meta(out, remote_index)
         stage.dump(with_files=True, update_pipeline=False)

dvc/schema.py

@@ -4,7 +4,13 @@ from voluptuous import Any, Optional, Required, Schema
 
 from dvc import dependency, output
 from dvc.annotations import ANNOTATION_SCHEMA
-from dvc.output import CHECKSUMS_SCHEMA, DIR_FILES_SCHEMA, META_SCHEMA, Output
+from dvc.output import (
+    CHECKSUMS_SCHEMA,
+    CLOUD_SCHEMA,
+    DIR_FILES_SCHEMA,
+    META_SCHEMA,
+    Output,
+)
 from dvc.parsing import DO_KWD, FOREACH_KWD, VARS_KWD
 from dvc.parsing.versions import SCHEMA_KWD, lockfile_version_schema
 from dvc.stage.params import StageParams
@@ -27,6 +33,7 @@ DATA_SCHEMA = {
     **CHECKSUMS_SCHEMA,
     **META_SCHEMA,
     Required("path"): str,
+    Output.PARAM_CLOUD: CLOUD_SCHEMA,
     Output.PARAM_FILES: [DIR_FILES_SCHEMA],
 }
 LOCK_FILE_STAGE_SCHEMA = {

dvc/scm.py

@@ -2,7 +2,16 @@
 import os
 from contextlib import contextmanager
 from functools import partial
-from typing import TYPE_CHECKING, Iterator, List, Mapping, Optional
+from typing import (
+    TYPE_CHECKING,
+    Iterator,
+    List,
+    Literal,
+    Mapping,
+    Optional,
+    Union,
+    overload,
+)
 
 from funcy import group_by
 from scmrepo.base import Base  # noqa: F401, pylint: disable=unused-import
@@ -81,8 +90,38 @@ def map_scm_exception(with_cause: bool = False) -> Iterator[None]:
         raise into
 
 
+@overload
 def SCM(
-    root_dir, search_parent_directories=True, no_scm=False
+    root_dir: str,
+    *,
+    search_parent_directories: bool = ...,
+    no_scm: Literal[False] = ...,
+) -> "Git":
+    ...
+
+
+@overload
+def SCM(
+    root_dir: str,
+    *,
+    search_parent_directories: bool = ...,
+    no_scm: Literal[True],
+) -> "NoSCM":
+    ...
+
+
+@overload
+def SCM(
+    root_dir: str,
+    *,
+    search_parent_directories: bool = ...,
+    no_scm: bool = ...,
+) -> Union["Git", "NoSCM"]:
+    ...
+
+
+def SCM(
+    root_dir, *, search_parent_directories=True, no_scm=False
 ):  # pylint: disable=invalid-name
     """Returns SCM instance that corresponds to a repo at the specified
     path.
@@ -242,9 +281,9 @@ def iter_revs(  # noqa: C901
 
             def _time_filter(rev):
                 try:
-                    if scm.resolve_commit(rev).commit_date >= commit_datestamp:
-                        return True
-                    return False
+                    return (
+                        scm.resolve_commit(rev).commit_time >= commit_datestamp
+                    )
                 except _SCMError:
                     return True
 

dvc/stage/__init__.py

@@ -486,17 +486,11 @@ class Stage(params.StageParams):
         logger.debug("Computed %s md5: '%s'", self, m)
         return m
 
-    def save(
-        self,
-        allow_missing: bool = False,
-        merge_versioned: bool = False,
-        run_cache: bool = True,
-    ):
+    def save(self, allow_missing: bool = False, run_cache: bool = True):
         self.save_deps(allow_missing=allow_missing)
 
-        self.save_outs(
-            allow_missing=allow_missing, merge_versioned=merge_versioned
-        )
+        self.save_outs(allow_missing=allow_missing)
+
         self.md5 = self.compute_md5()
 
         if run_cache:
@@ -512,29 +506,26 @@ class Stage(params.StageParams):
                 if not allow_missing:
                     raise
 
-    def save_outs(
-        self, allow_missing: bool = False, merge_versioned: bool = False
-    ):
+    def save_outs(self, allow_missing: bool = False):
         from dvc.output import OutputDoesNotExistError
 
         from .exceptions import StageFileDoesNotExistError, StageNotFound
 
-        if merge_versioned:
-            try:
-                old = self.reload()
-                old_outs = {out.def_path: out for out in old.outs}
-                merge_versioned = any(
-                    (
-                        out.files is not None
-                        or (
-                            out.meta is not None
-                            and out.meta.version_id is not None
-                        )
+        try:
+            old = self.reload()
+            old_outs = {out.def_path: out for out in old.outs}
+            merge_versioned = any(
+                (
+                    out.files is not None
+                    or (
+                        out.meta is not None
+                        and out.meta.version_id is not None
                     )
-                    for out in old_outs.values()
                 )
-            except (StageFileDoesNotExistError, StageNotFound):
-                merge_versioned = False
+                for out in old_outs.values()
+            )
+        except (StageFileDoesNotExistError, StageNotFound):
+            merge_versioned = False
 
         for out in self.outs:
             try:

dvc/stage/loader.py

@@ -2,7 +2,7 @@ import logging
 from collections.abc import Mapping
 from copy import deepcopy
 from itertools import chain
-from typing import TYPE_CHECKING, Any, Dict, Optional
+from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Tuple
 
 from funcy import get_in, lcat, once, project
 
@@ -20,6 +20,7 @@ from .utils import fill_stage_dependencies, resolve_paths
 
 if TYPE_CHECKING:
     from dvc.dvcfile import ProjectFile, SingleStageFile
+    from dvc.output import Output
 
 logger = logging.getLogger(__name__)
 
@@ -56,8 +57,10 @@ class StageLoader(Mapping):
         if not lock_data:
             return
 
+        from dvc.output import merge_file_meta_from_cloud
+
         assert isinstance(lock_data, dict)
-        items = chain(
+        items: Iterable[Tuple[str, "Output"]] = chain(
             ((StageParams.PARAM_DEPS, dep) for dep in stage.deps),
             ((StageParams.PARAM_OUTS, out) for out in stage.outs),
         )
@@ -74,19 +77,23 @@ class StageLoader(Mapping):
             info = get_in(checksums, [key, path], {})
             info = info.copy()
             info.pop("path", None)
-            item.meta = Meta.from_dict(info)
+
+            item.meta = Meta.from_dict(merge_file_meta_from_cloud(info))
             hash_value = getattr(item.meta, item.hash_name, None)
             item.hash_info = HashInfo(item.hash_name, hash_value)
-            if item.hash_info and item.hash_info.isdir:
-                item.meta.isdir = True
-            item.files = get_in(checksums, [key, path, item.PARAM_FILES])
+            files = get_in(checksums, [key, path, item.PARAM_FILES], None)
+            if files:
+                item.files = [merge_file_meta_from_cloud(f) for f in files]
+            # pylint: disable-next=protected-access
+            item._compute_meta_hash_info_from_files()
 
     @classmethod
     def load_stage(
         cls, dvcfile: "ProjectFile", name, stage_data, lock_data=None
     ):
         assert all([name, dvcfile, dvcfile.repo, dvcfile.path])
-        assert stage_data and isinstance(stage_data, dict)
+        assert stage_data
+        assert isinstance(stage_data, dict)
 
         path, wdir = resolve_paths(
             dvcfile.repo.fs, dvcfile.path, stage_data.get(Stage.PARAM_WDIR)

dvc/stage/monitor.py

@@ -110,7 +110,7 @@ class Monitor:
                     try:
                         task.execute()
                         task.updated.set()
-                    except Exception:  # noqa: BLE001, pylint: disable=W0703
+                    except Exception:  # pylint: disable=broad-except
                         logger.exception(
                             "Error running '%s' task, '%s' will be aborted",
                             task.name,

dvc/stage/run.py

@@ -22,7 +22,7 @@ def _make_cmd(executable, cmd):
         return cmd
     opts = {"zsh": ["--no-rcs"], "bash": ["--noprofile", "--norc"]}
     name = os.path.basename(executable).lower()
-    return [executable] + opts.get(name, []) + ["-c", cmd]
+    return [executable, *opts.get(name, []), "-c", cmd]
 
 
 def warn_if_fish(executable):

dvc/stage/serialize.py

@@ -2,11 +2,13 @@ from collections import OrderedDict
 from operator import attrgetter
 from typing import (
     TYPE_CHECKING,
+    Any,
     Dict,
     Iterable,
     List,
     Optional,
     Union,
+    cast,
     no_type_check,
 )
 
@@ -22,6 +24,7 @@ from .utils import resolve_wdir, split_params_deps
 
 if TYPE_CHECKING:
     from dvc.stage import PipelineStage, Stage
+    from dvc_data.hashfile.tree import Tree
 
 PARAM_PARAMS = ParamsDependency.PARAM_PARAMS
 PARAM_PATH = ParamsDependency.PARAM_PATH
@@ -153,25 +156,26 @@ def to_pipeline_file(stage: "PipelineStage"):
 
 
 def to_single_stage_lockfile(stage: "Stage", **kwargs) -> dict:
+    from dvc.output import split_file_meta_from_cloud
+
     assert stage.cmd
 
-    def _dumpd(item):
-        meta_d = item.meta.to_dict()
-        meta_d.pop("isdir", None)
-        ret = [
-            (item.PARAM_PATH, item.def_path),
-            *item.hash_info.to_dict().items(),
-            *meta_d.items(),
-        ]
-
+    def _dumpd(item: "Output"):
+        ret: Dict[str, Any] = {item.PARAM_PATH: item.def_path}
         if item.hash_info.isdir and kwargs.get("with_files"):
-            if item.obj:
-                obj = item.obj
-            else:
-                obj = item.get_obj()
-            ret.append((item.PARAM_FILES, obj.as_list(with_meta=True)))
-
-        return OrderedDict(ret)
+            obj = item.obj or item.get_obj()
+            if obj:
+                obj = cast("Tree", obj)
+                ret[item.PARAM_FILES] = [
+                    split_file_meta_from_cloud(f)
+                    for f in obj.as_list(with_meta=True)
+                ]
+        else:
+            meta_d = item.meta.to_dict()
+            meta_d.pop("isdir", None)
+            ret.update(item.hash_info.to_dict())
+            ret.update(split_file_meta_from_cloud(meta_d))
+        return ret
 
     res = OrderedDict([("cmd", stage.cmd)])
     params, deps = split_params_deps(stage)

dvc/testing/fixtures.py

@@ -12,11 +12,16 @@ __all__ = [
     "scm",
     "dvc",
     "make_cloud",
+    "make_cloud_version_aware",
     "make_local",
     "cloud",
     "local_cloud",
     "make_remote",
+    "make_remote_version_aware",
+    "make_remote_worktree",
     "remote",
+    "remote_version_aware",
+    "remote_worktree",
     "local_remote",
     "workspace",
     "make_workspace",
@@ -37,9 +42,8 @@ def make_tmp_dir(tmp_path_factory, request, worker_id):
     ):  # pylint: disable=W0621
         from shutil import copytree, ignore_patterns
 
-        from scmrepo.git import Git
-
         from dvc.repo import Repo
+        from dvc.scm import Git
 
         from .tmp_dir import TmpDir
 
@@ -104,6 +108,14 @@ def make_cloud(request):
     return _make_cloud
 
 
+@pytest.fixture
+def make_cloud_version_aware(request):
+    def _make_cloud(typ):
+        return request.getfixturevalue(f"make_{typ}_version_aware")()
+
+    return _make_cloud
+
+
 @pytest.fixture
 def cloud(make_cloud, request):
     typ = getattr(request, "param", "local")
@@ -125,12 +137,52 @@ def make_remote(tmp_dir, dvc, make_cloud):  # noqa: ARG001
     return _make_remote
 
 
+@pytest.fixture
+def make_remote_version_aware(
+    tmp_dir, dvc, make_cloud_version_aware  # noqa: ARG001
+):
+    def _make_remote(name, typ="local", **kwargs):
+        cloud = make_cloud_version_aware(typ)  # pylint: disable=W0621
+        config = dict(cloud.config)
+        config["version_aware"] = True
+        tmp_dir.add_remote(name=name, config=config, **kwargs)
+        return cloud
+
+    return _make_remote
+
+
+@pytest.fixture
+def make_remote_worktree(
+    tmp_dir, dvc, make_cloud_version_aware  # noqa: ARG001
+):
+    def _make_remote(name, typ="local", **kwargs):
+        cloud = make_cloud_version_aware(typ)  # pylint: disable=W0621
+        config = dict(cloud.config)
+        config["worktree"] = True
+        tmp_dir.add_remote(name=name, config=config, **kwargs)
+        return cloud
+
+    return _make_remote
+
+
 @pytest.fixture
 def remote(make_remote, request):
     typ = getattr(request, "param", "local")
     return make_remote("upstream", typ=typ)
 
 
+@pytest.fixture
+def remote_version_aware(make_remote_version_aware, request):
+    typ = getattr(request, "param", "local")
+    return make_remote_version_aware("upstream", typ=typ)
+
+
+@pytest.fixture
+def remote_worktree(make_remote_worktree, request):
+    typ = getattr(request, "param", "local")
+    return make_remote_worktree("upstream", typ=typ)
+
+
 @pytest.fixture
 def local_remote(make_remote):
     return make_remote("upstream", typ="local")

dvc/testing/path_info.py

@@ -129,7 +129,9 @@ class URLInfo(_BasePath):
 
     def __init__(self, url):
         p = urlparse(url)
-        assert not p.query and not p.params and not p.fragment
+        assert not p.query
+        assert not p.params
+        assert not p.fragment
         assert p.password is None
         self._fill_parts(p.scheme, p.hostname, p.username, p.port, p.path)
 

dvc/testing/remote_tests.py

@@ -112,3 +112,157 @@ class TestRemote:
 
         dvc.pull(run_cache=True)
         assert list(stage_cache_dir.rglob("*")) == expected
+
+
+class TestRemoteVersionAware:
+    def test_file(
+        self, tmp_dir, dvc, remote_version_aware  # pylint: disable=W0613
+    ):
+        (stage,) = tmp_dir.dvc_gen("foo", "foo")
+
+        dvc.push()
+        assert "version_id" in (tmp_dir / "foo.dvc").read_text()
+        stage = stage.reload()
+        out = stage.outs[0]
+        assert out.meta.version_id
+
+        remove(dvc.odb.local.path)
+        remove(tmp_dir / "foo")
+
+        dvc.pull()
+        assert (tmp_dir / "foo").read_text() == "foo"
+
+    def test_dir(
+        self, tmp_dir, dvc, remote_version_aware  # pylint: disable=W0613
+    ):
+        (stage,) = tmp_dir.dvc_gen(
+            {
+                "data_dir": {
+                    "data_sub_dir": {"data_sub": "data_sub"},
+                    "data": "data",
+                    "empty": "",
+                }
+            }
+        )
+
+        dvc.push()
+        assert "files" in (tmp_dir / "data_dir.dvc").read_text()
+        assert "version_id" in (tmp_dir / "data_dir.dvc").read_text()
+        stage = stage.reload()
+        out = stage.outs[0]
+        assert out.files
+        for file in out.files:
+            assert file["version_id"]
+            assert file["remote"] == "upstream"
+
+        remove(dvc.odb.local.path)
+        remove(tmp_dir / "data_dir")
+
+        dvc.pull()
+        assert (tmp_dir / "data_dir" / "data").read_text() == "data"
+        assert (
+            tmp_dir / "data_dir" / "data_sub_dir" / "data_sub"
+        ).read_text() == "data_sub"
+
+
+class TestRemoteWorktree:
+    def test_file(
+        self, tmp_dir, dvc, remote_worktree  # pylint: disable=W0613
+    ):
+        (stage,) = tmp_dir.dvc_gen("foo", "foo")
+
+        dvc.push()
+        assert "version_id" in (tmp_dir / "foo.dvc").read_text()
+        stage = stage.reload()
+        out = stage.outs[0]
+        assert out.meta.version_id
+
+        remove(dvc.odb.local.path)
+        remove(tmp_dir / "foo")
+
+        dvc.pull()
+        assert (tmp_dir / "foo").read_text() == "foo"
+
+    def test_dir(self, tmp_dir, dvc, remote_worktree):  # pylint: disable=W0613
+        (stage,) = tmp_dir.dvc_gen(
+            {
+                "data_dir": {
+                    "data_sub_dir": {"data_sub": "data_sub"},
+                    "data": "data",
+                    "empty": "",
+                }
+            }
+        )
+
+        dvc.push()
+        assert "files" in (tmp_dir / "data_dir.dvc").read_text()
+        assert "version_id" in (tmp_dir / "data_dir.dvc").read_text()
+        stage = stage.reload()
+        out = stage.outs[0]
+        assert out.files
+        for file in out.files:
+            assert file["version_id"]
+            assert file["remote"] == "upstream"
+
+        remove(dvc.odb.local.path)
+        remove(tmp_dir / "data_dir")
+
+        dvc.pull()
+        assert (tmp_dir / "data_dir" / "data").read_text() == "data"
+        assert (
+            tmp_dir / "data_dir" / "data_sub_dir" / "data_sub"
+        ).read_text() == "data_sub"
+
+    def test_deletion(
+        self, tmp_dir, dvc, scm, remote_worktree  # pylint: disable=W0613
+    ):
+        tmp_dir.dvc_gen(
+            {
+                "data_dir": {
+                    "data_sub_dir": {"data_sub": "data_sub"},
+                    "data": "data",
+                    "empty": "",
+                }
+            }
+        )
+        dvc.push()
+        assert (remote_worktree / "data_dir" / "data").exists()
+        tmp_dir.scm_add([tmp_dir / "data_dir.dvc"], commit="v1")
+        v1 = scm.get_rev()
+        remove(tmp_dir / "data_dir" / "data")
+        dvc.add(str(tmp_dir / "data_dir"))
+
+        # data_dir/data should show as deleted in the remote
+        dvc.push()
+        tmp_dir.scm_add([tmp_dir / "data_dir.dvc"], commit="v2")
+        assert not (remote_worktree / "data_dir" / "data").exists()
+
+        remove(dvc.odb.local.path)
+        remove(tmp_dir / "data_dir")
+        # pulling the original pushed version should still succeed
+        scm.checkout(v1)
+        dvc.pull()
+        assert (tmp_dir / "data_dir" / "data").read_text() == "data"
+
+    def test_update(
+        self, tmp_dir, dvc, remote_worktree  # pylint: disable=W0613
+    ):
+        tmp_dir.dvc_gen("foo", "foo")
+        tmp_dir.dvc_gen(
+            {
+                "data_dir": {
+                    "data_sub_dir": {"data_sub": "data_sub"},
+                    "data": "data",
+                    "empty": "",
+                }
+            }
+        )
+        dvc.push()
+        (remote_worktree / "foo").write_text("bar")
+        (remote_worktree / "data_dir" / "data").write_text("modified")
+        (remote_worktree / "data_dir" / "new_data").write_text("new data")
+
+        dvc.update([str(tmp_dir / "foo.dvc"), str(tmp_dir / "data_dir.dvc")])
+        assert (tmp_dir / "foo").read_text() == "bar"
+        assert (tmp_dir / "data_dir" / "data").read_text() == "modified"
+        assert (tmp_dir / "data_dir" / "new_data").read_text() == "new data"

dvc/testing/tmp_dir.py

@@ -87,9 +87,8 @@ class TmpDir(pathlib.Path):
         return self
 
     def init(self, *, scm=False, dvc=False, subdir=False):
-        from scmrepo.git import Git
-
         from dvc.repo import Repo
+        from dvc.scm import Git
 
         assert not scm or not hasattr(self, "scm")
         assert not dvc or not hasattr(self, "dvc")
@@ -158,7 +157,7 @@ class TmpDir(pathlib.Path):
 
     def commit(self, output_paths, msg):
         def to_gitignore(stage_path):
-            from scmrepo.git import Git
+            from dvc.scm import Git
 
             return os.path.join(os.path.dirname(stage_path), Git.GITIGNORE)
 

dvc/testing/workspace_tests.py

@@ -2,6 +2,7 @@ import os
 from typing import Dict, Union
 
 import pytest
+from funcy import first
 
 from dvc.exceptions import URLMissingError
 from dvc.repo import Repo
@@ -88,6 +89,42 @@ class TestImport:
         assert tuple(empty_dir.iterdir()) == ()
 
 
+class TestImportURLVersionAware:
+    def test_import_file(self, tmp_dir, dvc, remote_version_aware):
+        remote_version_aware.gen("file", "file")
+        dvc.imp_url("remote://upstream/file", version_aware=True)
+        stage = first(dvc.index.stages)
+        assert not stage.outs[0].can_push
+        assert (tmp_dir / "file").read_text() == "file"
+        assert dvc.status() == {}
+
+        (remote_version_aware / "file").write_text("modified")
+        assert dvc.status().get("file.dvc") == [
+            {"changed deps": {"remote://upstream/file": "update available"}},
+        ]
+        dvc.update(str(tmp_dir / "file.dvc"))
+        assert (tmp_dir / "file").read_text() == "modified"
+        assert dvc.status() == {}
+
+    def test_import_dir(self, tmp_dir, dvc, remote_version_aware):
+        remote_version_aware.gen({"data_dir": {"file": "file"}})
+        dvc.imp_url("remote://upstream/data_dir", version_aware=True)
+        stage = first(dvc.index.stages)
+        assert not stage.outs[0].can_push
+        assert (tmp_dir / "data_dir" / "file").read_text() == "file"
+        assert dvc.status() == {}
+
+        (remote_version_aware / "data_dir" / "file").write_text("modified")
+        (remote_version_aware / "data_dir" / "new_file").write_text("new")
+        assert dvc.status().get("data_dir.dvc") == [
+            {"changed deps": {"remote://upstream/data_dir": "modified"}},
+        ]
+        dvc.update(str(tmp_dir / "data_dir.dvc"))
+        assert (tmp_dir / "data_dir" / "file").read_text() == "modified"
+        assert (tmp_dir / "data_dir" / "new_file").read_text() == "new"
+        assert dvc.status() == {}
+
+
 class TestAdd:
     @pytest.fixture
     def hash_name(self):

dvc/types.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, Any, AnyStr, Dict, List, Optional, Union
+from typing import TYPE_CHECKING, Any, AnyStr, Dict, List, Union
 
 if TYPE_CHECKING:
     from os import PathLike
@@ -8,7 +8,5 @@ BytesPath = Union[bytes, "PathLike[bytes]"]
 GenericPath = Union[AnyStr, "PathLike[AnyStr]"]
 StrOrBytesPath = Union[str, bytes, "PathLike[str]", "PathLike[bytes]"]
 
-OptStr = Optional[str]
 TargetType = Union[List[str], str]
 DictStrAny = Dict[str, Any]
-DictAny = Dict[Any, Any]

dvc/ui/__init__.py

@@ -38,7 +38,7 @@ class Formatter:
         }
         self.theme = defaultdict(lambda: defaults or {}, theme)
 
-    def format(
+    def format(  # noqa: A003
         self, message: str, style: Optional[str] = None, **kwargs
     ) -> str:
         from dvc.utils import colorize

dvc/utils/collections.py

@@ -53,7 +53,7 @@ def apply_diff(src, dest):  # noqa: C901
                 elif value != dest[i]:
                     dest[i] = value
     else:
-        raise AssertionError(
+        raise AssertionError(  # noqa: TRY004
             "Can't apply diff from {} to {}".format(
                 src.__class__.__name__, dest.__class__.__name__
             )

dvc/utils/strictyaml.py

@@ -93,7 +93,7 @@ class YAMLSyntaxError(PrettyDvcException, YAMLFileCorruptedError):
         exc = self.exc.__cause__
 
         if not isinstance(exc, MarkedYAMLError):
-            raise ValueError("nothing to pretty-print here. :)")
+            raise ValueError("nothing to pretty-print here.")  # noqa: TRY004
 
         source = self.yaml_text.splitlines()
 

pyproject.toml

@@ -49,14 +49,14 @@ dependencies = [
     "rich>=10.13.0",
     "pyparsing>=2.4.7",
     "typing-extensions>=3.7.4",
-    "scmrepo==0.1.6",
-    "dvc-render==0.0.17",
+    "scmrepo==0.1.7",
+    "dvc-render==0.1.0",
     "dvc-task==0.1.11",
     "dvclive>=1.2.2",
-    "dvc-data==0.35.1",
+    "dvc-data==0.37.3",
     "dvc-http",
     "hydra-core>=1.1.0",
-    "iterative-telemetry==0.0.6",
+    "iterative-telemetry==0.0.7",
     "dvc-studio-client>=0.1.1"
 ]
 dynamic = ["version"]
@@ -72,13 +72,13 @@ classifiers = [
 [project.optional-dependencies]
 azure = ["dvc-azure==2.21.0"]
 gdrive = ["dvc-gdrive==2.19.1"]
-gs = ["dvc-gs==2.20.0"]
+gs = ["dvc-gs==2.21.1"]
 hdfs = ["dvc-hdfs==2.19.0"]
 oss = ["dvc-oss==2.19.0"]
 s3 = ["dvc-s3==2.21.0"]
 ssh = ["dvc-ssh==2.21.0"]
 ssh_gssapi = ["dvc-ssh[gssapi]==2.21.0"]
-webdav = ["dvc-webdav==2.19.0"]
+webdav = ["dvc-webdav==2.19.1"]
 webhdfs = ["dvc-webhdfs==2.19.0"]
 webhdfs_kerberos = ["dvc-webhdfs[kerberos]==2.19.0"]
 terraform = ["tpi[ssh]>=2.1.0"]
@@ -98,9 +98,9 @@ tests = [
     "flaky==3.7.0",
     "pytest-timeout==2.1.0",
     "filelock==3.9.0",
-    "beautifulsoup4==4.11.1",
+    "beautifulsoup4==4.11.2",
     "pywin32>=225; sys_platform == 'win32'",  # optional dependency
-    "pylint==2.15.10",
+    "pylint==2.16.0",
     # type-checking
     "mypy==0.991",
     "types-requests",
@@ -161,26 +161,32 @@ line_length = 79
 line-length = 79
 target-version = "py38"
 show-source = true
-ignore = ["S101", "SIM105", "SIM108", "SIM110", "SIM111", "SIM117", "A003", "ARG002", "PLR2004"]
-select = ["F", "E", "W", "C90", "UP", "YTT", "S", "BLE", "B", "A", "C4", "T10", "ISC", "ICN", "T20", "PT003", "PT008", "PT009", "PT010", "PT015", "PT016", "PT017", "PT020", "PT021", "PT024", "PT025", "PT026", "Q", "RET501", "RET504", "SIM", "ARG", "TID", "PGH", "PLC", "PLE", "PLR", "PLW", "PIE", "RUF"]
+ignore = ["S101", "PT004", "PT007", "PT019", "SIM105", "SIM108", "SIM110", "SIM111", "SIM117", "TRY003", "TRY200", "TRY300", "TRY301", "PLR0913", "PLR2004", "RUF005"]
+select = ["F", "E", "W", "C90", "UP", "YTT", "S", "BLE", "B", "A", "C4", "T10", "ISC", "ICN", "G", "INP", "PIE", "T20", "PT", "Q", "RET501", "RET504", "SIM", "TID", "TCH", "ARG", "PGH", "PLC", "PLE", "PLR", "PLW", "TRY", "RUF"]
 # external flake8 codes that should be preserved
 external = ["B301", "B024", "C901", "E302", "F401", "W601"]
 
-[tool.ruff.per-file-ignores]
-"scripts/**/*.py" = ["T201"]
-
 [tool.ruff.flake8-pytest-style]
 fixture-parentheses = false
 mark-parentheses = false
 parametrize-names-type = "csv"
-
-[tool.ruff.flake8-unused-arguments]
-ignore-variadic-names = true
+raises-extend-require-match-for = ["dvc.exceptions.DvcException", "dvc.scm.SCMError", "scmrepo.exceptions.SCMError"]
 
 [tool.ruff.flake8-tidy-imports]
 [tool.ruff.flake8-tidy-imports.banned-api]
 "funcy.cached_property".msg = "use `from dvc.utils.objects import cached_property` instead."
 
+[tool.ruff.flake8-type-checking]
+exempt-modules = ["typing"]
+strict = true
+
+[tool.ruff.flake8-unused-arguments]
+ignore-variadic-names = true
+
+[tool.ruff.per-file-ignores]
+"dvc/testing/*.py" = ["ARG002"]
+"scripts/**/*.py" = ["T201", "INP001", "TRY002"]
+
 [tool.pytest.ini_options]
 log_level = "debug"
 addopts = "-ra --cov-config pyproject.toml"
@@ -236,16 +242,11 @@ module = [
     "celery.*",
     "configobj.*",
     "dpath.*",
-    "dvc_http.*",
-    "dvc_render.*",
-    "dvc_ssh",
-    "dvc_studio_client.*",
     "flatten_dict",
     "fsspec.*",
     "funcy.*",  # https://github.com/Suor/funcy/issues/106,
     "grandalf.*",
     "ipdb",
-    "iterative_telemetry",
     "kombu.*",
     "networkx.*",  # https://github.com/networkx/networkx/issues/3988
     "pygtrie.*",
@@ -253,7 +254,6 @@ module = [
     "pytest_docker.plugin",
     "ruamel.*",
     "ruamel.yaml.*",
-    "scmrepo.*",
     "tpi.*",
     "viztracer",
     "voluptuous",
@@ -264,11 +264,9 @@ ignore_missing_imports = true
 
 [tool.pylint.message_control]
 disable = [
-    "format", "refactoring", "spelling", "design",
-    "invalid-name", "duplicate-code", "fixme",
-    "unused-wildcard-import", "cyclic-import", "wrong-import-order",
+    "format", "refactoring", "spelling", "design", "invalid-name", "fixme",
+    "duplicate-code", "unused-wildcard-import", "cyclic-import", "wrong-import-order",
     "wrong-import-position", "ungrouped-imports", "multiple-imports",
-    "logging-format-interpolation", "logging-fstring-interpolation",
     "missing-function-docstring", "missing-module-docstring",
     "missing-class-docstring", "raise-missing-from", "import-outside-toplevel",
 ]

tests/func/experiments/test_checkpoints.py

@@ -10,7 +10,7 @@ from dvc.repo.experiments.exceptions import MultipleBranchError
 from dvc.repo.experiments.executor.base import BaseExecutor
 from dvc.repo.experiments.refs import EXEC_APPLY, EXEC_CHECKPOINT
 from dvc.repo.experiments.utils import exp_refs_by_rev
-from dvc.scm import InvalidRemoteSCMRepo
+from dvc.scm import InvalidRemoteSCMRepo, RevError
 
 
 @pytest.mark.parametrize("links", ["reflink,copy", "hardlink,symlink"])
@@ -54,7 +54,7 @@ def test_resume_checkpoint(
         checkpoint_stage.addressing, params=["foo=2"], tmp_dir=not workspace
     )
 
-    with pytest.raises(DvcException):
+    with pytest.raises(RevError, match="unknown Git revision 'abc1234'"):
         dvc.experiments.run(
             checkpoint_stage.addressing,
             checkpoint_resume="abc1234",
@@ -177,7 +177,9 @@ def test_resume_non_head_checkpoint(
     rev = list(scm.branch_revs(checkpoint_head, orig_head))[-1]
     dvc.experiments.apply(rev)
 
-    with pytest.raises(DvcException):
+    with pytest.raises(
+        DvcException, match="Nothing to do for unchanged checkpoint"
+    ):
         dvc.experiments.run(checkpoint_stage.addressing, tmp_dir=not workspace)
 
     results = dvc.experiments.run(

tests/func/experiments/test_diff.py

@@ -2,9 +2,8 @@ from funcy import first
 
 
 def test_diff_no_commits(tmp_dir):
-    from scmrepo.git import Git
-
     from dvc.repo import Repo
+    from dvc.scm import Git
 
     git = Git.init(tmp_dir.fs_path)
     assert git.no_commits

tests/func/experiments/test_experiments.py

@@ -32,7 +32,8 @@ def test_new_simple(tmp_dir, scm, dvc, exp_stage, mocker, name, workspace):
     )
     exp = first(results)
     ref_info = first(exp_refs_by_rev(scm, exp))
-    assert ref_info and ref_info.baseline_sha == baseline
+    assert ref_info
+    assert ref_info.baseline_sha == baseline
 
     new_mock.assert_called_once()
     fs = scm.get_fs(exp)
@@ -131,8 +132,8 @@ def test_apply(tmp_dir, scm, dvc, exp_stage):
     with pytest.raises(ApplyConflictError):
         dvc.experiments.apply(exp_b, force=False)
         # failed apply should revert everything to prior state
-        assert (tmp_dir / "params.yaml").read_text().strip() == "foo: 2"
-        assert (tmp_dir / "metrics.yaml").read_text().strip() == "foo: 2"
+    assert (tmp_dir / "params.yaml").read_text().strip() == "foo: 2"
+    assert (tmp_dir / "metrics.yaml").read_text().strip() == "foo: 2"
 
     dvc.experiments.apply("foo")
     assert (tmp_dir / "params.yaml").read_text().strip() == "foo: 3"

tests/func/experiments/test_save.py

@@ -13,7 +13,6 @@ from dvc.scm import resolve_rev
 def modified_exp_stage(exp_stage, tmp_dir):
     with open(tmp_dir / "copy.py", "a", encoding="utf-8") as fh:
         fh.write("\n# dummy change")
-    yield
 
 
 def test_exp_save_unchanged(tmp_dir, dvc, scm, exp_stage):
@@ -26,7 +25,8 @@ def test_exp_save(tmp_dir, dvc, scm, exp_stage, name, modified_exp_stage):
 
     exp = dvc.experiments.save(name=name)
     ref_info = first(exp_refs_by_rev(scm, exp))
-    assert ref_info and ref_info.baseline_sha == baseline
+    assert ref_info
+    assert ref_info.baseline_sha == baseline
 
     exp_name = name if name else ref_info.name
     assert dvc.experiments.get_exact_name([exp])[exp] == exp_name

tests/func/experiments/test_show.py

@@ -50,9 +50,8 @@ def make_executor_info(**kwargs):
 
 @pytest.mark.vscode
 def test_show_no_commits(tmp_dir):
-    from scmrepo.git import Git
-
     from dvc.repo import Repo
+    from dvc.scm import Git
 
     git = Git.init(tmp_dir.fs_path)
     assert git.no_commits

tests/func/machine/conftest.py

@@ -35,7 +35,7 @@ TEST_INSTANCE = {
 @pytest.fixture
 def machine_config(tmp_dir):
     (tmp_dir / ".dvc" / "config").write_text(BASIC_CONFIG)
-    yield BASIC_CONFIG
+    return BASIC_CONFIG
 
 
 @pytest.fixture
@@ -52,4 +52,4 @@ def machine_instance(tmp_dir, dvc, mocker):
         "tpi.terraform.TerraformBackend.instances",
         mocker.MagicMock(side_effect=mock_instances),
     )
-    yield TEST_INSTANCE
+    return TEST_INSTANCE

tests/func/metrics/test_diff.py

@@ -149,9 +149,8 @@ def test_metrics_diff_with_unchanged(tmp_dir, scm, dvc, run_copy_metrics):
 
 
 def test_no_commits(tmp_dir):
-    from scmrepo.git import Git
-
     from dvc.repo import Repo
+    from dvc.scm import Git
 
     git = Git.init(tmp_dir.fs_path)
     assert git.no_commits

tests/func/params/test_diff.py

@@ -130,9 +130,8 @@ def test_pipeline_tracked_params(tmp_dir, scm, dvc, run_copy):
 
 
 def test_no_commits(tmp_dir):
-    from scmrepo.git import Git
-
     from dvc.repo import Repo
+    from dvc.scm import Git
 
     git = Git.init(tmp_dir.fs_path)
     assert git.no_commits

tests/func/parsing/test_resolver.py

@@ -163,10 +163,9 @@ def test_vars_already_loaded_message(tmp_dir, dvc, local, vars_):
     else:
         d["stages"]["build"]["vars"] = vars_
 
-    with pytest.raises(ResolveError) as exc_info:
+    with pytest.raises(ResolveError) as exc_info:  # noqa: PT012
         resolver = DataResolver(dvc, tmp_dir.fs_path, d)
         resolver.resolve()
-
     assert "partially" in str(exc_info.value)
 
 

tests/func/plots/test_diff.py

@@ -52,9 +52,8 @@ def test_diff_dirty(tmp_dir, scm, dvc, run_copy_metrics):
 
 @pytest.mark.vscode
 def test_no_commits(tmp_dir):
-    from scmrepo.git import Git
-
     from dvc.repo import Repo
+    from dvc.scm import Git
 
     git = Git.init(tmp_dir.fs_path)
     assert git.no_commits

tests/func/test_add.py

@@ -13,6 +13,7 @@ import dvc as dvc_module
 import dvc_data
 from dvc.annotations import Annotation
 from dvc.cli import main
+from dvc.config import ConfigError
 from dvc.dvcfile import DVC_FILE_SUFFIX
 from dvc.exceptions import (
     DvcException,
@@ -97,7 +98,7 @@ def test_add_unicode(tmp_dir, dvc):
 
 
 def test_add_unsupported_file(dvc):
-    with pytest.raises(DvcException):
+    with pytest.raises(ConfigError, match="Unsupported URL type"):
         dvc.add("unsupported://unsupported")
 
 
@@ -810,7 +811,8 @@ def test_add_symlink_dir(make_tmp_dir, tmp_dir, dvc, external):
 
     (tmp_dir / "dir").symlink_to(target)
 
-    with pytest.raises(DvcException):
+    msg = "Cannot add files inside symlinked directories to DVC"
+    with pytest.raises(DvcException, match=msg):
         dvc.add("dir")
 
 
@@ -826,7 +828,8 @@ def test_add_file_in_symlink_dir(make_tmp_dir, tmp_dir, dvc, external):
 
     (tmp_dir / "dir").symlink_to(target)
 
-    with pytest.raises(DvcException):
+    msg = "Cannot add files inside symlinked directories to DVC"
+    with pytest.raises(DvcException, match=msg):
         dvc.add(os.path.join("dir", "foo"))
 
 
@@ -895,7 +898,7 @@ def test_add_long_fname(tmp_dir, dvc):
 
     # nothing we can do in this case, as the resulting dvcfile
     # will definitely exceed NAME_MAX
-    with pytest.raises(OSError) as info:
+    with pytest.raises(OSError, match=f"File name too long: .*{name}") as info:
         dvc.add(os.path.join("data", name))
     assert info.value.errno == errno.ENAMETOOLONG
 
@@ -939,8 +942,8 @@ def test_add_to_remote_absolute(tmp_dir, make_tmp_dir, dvc, remote):
     assert not os.path.exists(tmp_foo)
     assert foo.read_text() == "foo"
 
+    tmp_bar = tmp_abs_dir / "bar"
     with pytest.raises(StageExternalOutputsError):
-        tmp_bar = tmp_abs_dir / "bar"
         dvc.add(str(tmp_foo), out=str(tmp_bar), to_remote=True)
 
 
@@ -1108,9 +1111,8 @@ def test_add_does_not_remove_stage_file_on_failure(
         side_effect=DvcException(exc_msg),
     )
 
-    with pytest.raises(DvcException) as exc_info:
+    with pytest.raises(DvcException, match=exc_msg):
         dvc.add("foo")
-    assert str(exc_info.value) == exc_msg
     assert (tmp_dir / "foo.dvc").exists()
     assert (tmp_dir / stage.path).read_text() == dvcfile_contents
 
@@ -1142,3 +1144,61 @@ def test_add_with_annotations(M, tmp_dir, dvc):
     (stage,) = dvc.add("foo", type="t2")
     assert stage.outs[0].annot == Annotation(**annot)
     assert (tmp_dir / "foo.dvc").parse() == M.dict(outs=[M.dict(**annot)])
+
+
+def test_add_updates_to_cloud_versioning_dir(tmp_dir, dvc):
+    data_dvc = tmp_dir / "data.dvc"
+    data_dvc.dump(
+        {
+            "outs": [
+                {
+                    "path": "data",
+                    "files": [
+                        {
+                            "size": 3,
+                            "version_id": "WYRG4BglP7pD.gEoJP6a4AqOhl.FRA.h",
+                            "etag": "acbd18db4cc2f85cedef654fccc4a4d8",
+                            "md5": "acbd18db4cc2f85cedef654fccc4a4d8",
+                            "relpath": "bar",
+                        },
+                        {
+                            "size": 3,
+                            "version_id": "0vL53tFVY5vVAoJ4HG2jCS1mEcohDPE0",
+                            "etag": "acbd18db4cc2f85cedef654fccc4a4d8",
+                            "md5": "acbd18db4cc2f85cedef654fccc4a4d8",
+                            "relpath": "foo",
+                        },
+                    ],
+                }
+            ]
+        }
+    )
+
+    data = tmp_dir / "data"
+    data.mkdir()
+    (data / "foo").write_text("foo")
+    (data / "bar").write_text("bar2")
+
+    dvc.add("data")
+
+    assert (tmp_dir / "data.dvc").parse() == {
+        "outs": [
+            {
+                "path": "data",
+                "files": [
+                    {
+                        "size": 4,
+                        "md5": "224e2539f52203eb33728acd228b4432",
+                        "relpath": "bar",
+                    },
+                    {
+                        "size": 3,
+                        "version_id": "0vL53tFVY5vVAoJ4HG2jCS1mEcohDPE0",
+                        "etag": "acbd18db4cc2f85cedef654fccc4a4d8",
+                        "md5": "acbd18db4cc2f85cedef654fccc4a4d8",
+                        "relpath": "foo",
+                    },
+                ],
+            }
+        ]
+    }

tests/func/test_analytics.py

@@ -31,8 +31,9 @@ def mock_daemon(mocker):
     def func(argv):
         return main(["daemon", *argv])
 
-    m = mocker.patch("dvc.daemon.daemon", mocker.MagicMock(side_effect=func))
-    yield m
+    return mocker.patch(
+        "dvc.daemon.daemon", mocker.MagicMock(side_effect=func)
+    )
 
 
 def test_collect_and_send_report(mocker, dvc, mock_daemon):

tests/func/test_check_ignore.py

@@ -65,7 +65,7 @@ def test_check_ignore_non_matching(tmp_dir, dvc, non_matching, caplog, capsys):
     ],
 )
 def test_check_ignore_error_args_cases(tmp_dir, dvc, args, caplog):
-    assert main(["check-ignore"] + args) == 255
+    assert main(["check-ignore", *args]) == 255
     assert "Having any troubles?" not in caplog.text
 
 

tests/func/test_cli.py

@@ -13,7 +13,7 @@ from dvc.commands.remove import CmdRemove
 from dvc.commands.repro import CmdRepro
 from dvc.commands.run import CmdRun
 from dvc.commands.status import CmdDataStatus
-from dvc.exceptions import DvcException
+from dvc.exceptions import NotDvcRepoError
 
 
 def test_argparse(dvc):
@@ -176,7 +176,7 @@ def test_find_root(dvc):
         cd = os.path.pardir
 
     args = A()
-    with pytest.raises(DvcException):
+    with pytest.raises(NotDvcRepoError):
         Cmd(args)
 
 
@@ -192,7 +192,7 @@ def test_cd(dvc):
 
     parent_dir = os.path.realpath(os.path.pardir)
     args = A()
-    with pytest.raises(DvcException):
+    with pytest.raises(NotDvcRepoError):
         Cmd(args)
     current_dir = os.path.realpath(os.path.curdir)
     assert parent_dir == current_dir