output: add progress for commit

Fixes #9656

.flake8

@@ -1,19 +1,20 @@
 [flake8]
-ignore=
+max_line_length = 88
+extend-ignore=
     # Whitespace before ':'
     E203
-    # Too many leading '#' for block comment
-    E266
     # Line break occurred before a binary operator
     W503
     # unindexed parameters in the str.format, see:
     # https://pypi.org/project/flake8-string-format/
     P1
-    # use `!r` instead of surrounding by quotes
-    B028
-max_line_length = 88
 max-complexity = 10
-select = B,C,E,F,W,T4,B902,T,P,TYP
 show_source = true
 count = true
 min_python_version = 3.8.0
+per-file-ignores =
+    dvc/testing/benchmarks/conftest.py:F401
+    dvc/testing/conftest.py:F401
+    dvc/testing/plugin.py:F401
+    tests/conftest.py:F401
+    tests/func/experiments/executor/test_ssh.py:F401

.github/workflows/build.yaml

@@ -1,4 +1,4 @@
-name: Build packages
+name: Build package
 on:
   push:
     branches:
@@ -7,39 +7,35 @@ on:
     types: [released, prereleased]
   workflow_dispatch:
 
-permissions:  # added using https://github.com/step-security/secure-workflows
+permissions:
   contents: read
+  id-token: write
 
 jobs:
   pip:
-    runs-on: ubuntu-20.04
-    permissions:
-      id-token: write
+    runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v3
       with:
         fetch-depth: 0
 
-    - name: Set up Python 3.8
-      uses: actions/setup-python@v4
+    - uses: actions/setup-python@v4
       with:
         python-version: 3.8
 
-    - name: Install
-      run: pip install --upgrade pip wheel
-
-    - name: Force version for Test PyPI uploads
+    - run: python -m pip install --upgrade pip setuptools_scm build twine
+    - name: Force version for Test PyPI upload
       if: ${{ !startsWith(github.ref, 'refs/tags') }}
-      run: |
-        pip install setuptools_scm
-        echo version=$(python -m setuptools_scm | awk -F+ '{print $1}' | tail -1) >> $GITHUB_ENV
+      run: echo version=$(python -m setuptools_scm | awk -F+ '{print $1}' | tail -1) >> $GITHUB_ENV
 
-    - name: Build packages
-      run: ./scripts/build_package.sh
+    - run: |
+        echo 'PKG = "pip"'>dvc/_build.py
+        python -m build
       env:
         SETUPTOOLS_SCM_PRETEND_VERSION: ${{ env.version }}
+    - run: twine check --strict dist/*
 
-    - name: Publish packages to PyPI
+    - name: Publish to PyPI
       if: github.event_name == 'release'
       uses: pypa/gh-action-pypi-publish@release/v1
 
@@ -47,5 +43,5 @@ jobs:
       if: ${{ github.event_name == 'release' || (github.event_name == 'push' && github.ref == 'refs/heads/main') }}
       uses: pypa/gh-action-pypi-publish@release/v1
       with:
-        repository_url: https://test.pypi.org/legacy/
-        skip_existing: true
+        repository-url: https://test.pypi.org/legacy/
+        skip-existing: true

.pre-commit-config.yaml

@@ -23,42 +23,29 @@ repos:
       - id: mixed-line-ending
       - id: sort-simple-yaml
       - id: trailing-whitespace
-  - repo: https://github.com/lovesegfault/beautysh
-    rev: v6.2.1
-    hooks:
-      - id: beautysh
-        args: ["-i", "2"]
-  - repo: https://github.com/asottile/pyupgrade
-    rev: v3.3.1
-    hooks:
-      - id: pyupgrade
-        args:
-          - --py38-plus
   - repo: https://github.com/psf/black
-    rev: 23.3.0
+    rev: 23.7.0
     hooks:
       - id: black
   - repo: https://github.com/PyCQA/isort
     rev: 5.12.0
     hooks:
       - id: isort
-  - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: 'v0.0.261'
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: 'v0.0.282'
     hooks:
       - id: ruff
+        args: [--fix, --exit-non-zero-on-fix]
   - repo: https://github.com/codespell-project/codespell
-    rev: v2.2.4
+    rev: v2.2.5
     hooks:
       - id: codespell
         additional_dependencies: ["tomli"]
   - repo: https://github.com/PyCQA/flake8
-    rev: 6.0.0
+    rev: 6.1.0
     hooks:
       - id: flake8
         additional_dependencies:
-          - flake8-bugbear
-          - flake8-comprehensions
-          - flake8-debugger
           - flake8-string-format
           - flake8-typing-imports
   - repo: https://github.com/PyCQA/bandit

README.rst

@@ -280,8 +280,8 @@ Barrak, A., Eghan, E.E. and Adams, B. `On the Co-evolution of ML Pipelines and S
    :target: https://pypi.org/project/dvc
    :alt: PyPI Downloads
 
-.. |Packages| image:: https://img.shields.io/github/v/release/iterative/dvc?label=deb|pkg|rpm|exe&logo=GitHub
-   :target: https://github.com/iterative/dvc/releases/latest
+.. |Packages| image:: https://img.shields.io/badge/deb|pkg|rpm|exe-blue
+   :target: https://dvc.org/doc/install
    :alt: deb|pkg|rpm|exe
 
 .. |DOI| image:: https://img.shields.io/badge/DOI-10.5281/zenodo.3677553-blue.svg

dvc/__pyinstaller/__init__.py

@@ -0,0 +1,5 @@
+import os
+
+
+def get_hook_dirs():
+    return [os.path.dirname(__file__)]

dvc/__pyinstaller/hook-asyncssh.py

@@ -1 +1,2 @@
+# ruff: noqa: N999
 hiddenimports = ["win32timezone"]

dvc/__pyinstaller/hook-celery.py

@@ -1,3 +1,4 @@
+# ruff: noqa: N999
 # pylint: disable=import-error
 from PyInstaller.utils.hooks import collect_submodules, is_module_or_submodule
 

dvc/__pyinstaller/hook-dvc.py

@@ -1,3 +1,4 @@
+# ruff: noqa: N999
 from PyInstaller.utils.hooks import copy_metadata  # pylint:disable=import-error
 
 # needed for `dvc doctor` to show dep versions

dvc/__pyinstaller/hook-dvc.utils.flatten.py

@@ -1,3 +1,4 @@
+# ruff: noqa: N999
 from PyInstaller.utils.hooks import copy_metadata  # pylint:disable=import-error
 
 datas = copy_metadata("flatten-dict")

dvc/__pyinstaller/hook-dvc_task.py

@@ -1,3 +1,4 @@
+# ruff: noqa: N999
 # pylint: disable=import-error
 from PyInstaller.utils.hooks import collect_submodules
 

dvc/__pyinstaller/hook-fsspec.py

@@ -1 +1,2 @@
+# ruff: noqa: N999
 hiddenimports = ["fsspec.implementations.memory"]

dvc/__pyinstaller/hook-pydrive2.py

@@ -1,3 +1,4 @@
+# ruff: noqa: N999
 from PyInstaller.utils.hooks import copy_metadata  # pylint:disable=import-error
 
 datas = copy_metadata("pydrive2")

dvc/_debug.py

@@ -198,33 +198,84 @@ def debugtools(args: Optional["Namespace"] = None, **kwargs):
 def add_debugging_flags(parser):
     from argparse import SUPPRESS
 
-    parser.add_argument("--cprofile", action="store_true", default=False, help=SUPPRESS)
-    parser.add_argument("--yappi", action="store_true", default=False, help=SUPPRESS)
+    # For detailed info see:
+    # https://github.com/iterative/dvc/wiki/Debugging,-Profiling-and-Benchmarking-DVC
+    args, _ = parser.parse_known_args()
+    verbose = args.verbose
+
+    def debug_help(msg):
+        if verbose:
+            return msg
+        return SUPPRESS
+
+    parser = parser.add_argument_group("debug options")
+
+    parser.add_argument(
+        "--cprofile",
+        action="store_true",
+        default=False,
+        help=debug_help("Generate cprofile data for tools like snakeviz / tuna"),
+    )
+    parser.add_argument(
+        "--cprofile-dump", help=debug_help("Location to dump cprofile file")
+    )
+    parser.add_argument(
+        "--yappi",
+        action="store_true",
+        default=False,
+        help=debug_help(
+            "Generate a callgrind file for use with tools like "
+            "kcachegrind / qcachegrind"
+        ),
+    )
     parser.add_argument(
         "--yappi-separate-threads",
         action="store_true",
         default=False,
-        help=SUPPRESS,
+        help=debug_help("Generate one callgrind file per thread"),
     )
     parser.add_argument(
-        "--viztracer", action="store_true", default=False, help=SUPPRESS
-    )
-    parser.add_argument("--viztracer-depth", type=int, help=SUPPRESS)
-    parser.add_argument(
-        "--viztracer-async", action="store_true", default=False, help=SUPPRESS
-    )
-    parser.add_argument("--cprofile-dump", help=SUPPRESS)
-    parser.add_argument("--pdb", action="store_true", default=False, help=SUPPRESS)
-    parser.add_argument(
-        "--instrument", action="store_true", default=False, help=SUPPRESS
+        "--viztracer",
+        action="store_true",
+        default=False,
+        help=debug_help("Generate a viztracer file for use with vizviewer"),
     )
     parser.add_argument(
-        "--instrument-open", action="store_true", default=False, help=SUPPRESS
+        "--viztracer-depth",
+        type=int,
+        help=debug_help("Set viztracer maximum stack depth"),
+    )
+    parser.add_argument(
+        "--viztracer-async",
+        action="store_true",
+        default=False,
+        help=debug_help("Treat async tasks as threads"),
+    )
+    parser.add_argument(
+        "--pdb",
+        action="store_true",
+        default=False,
+        help=debug_help("Drop into the pdb/ipdb debugger on any exception"),
+    )
+    parser.add_argument(
+        "--instrument",
+        action="store_true",
+        default=False,
+        help=debug_help("Use pyinstrument CLI profiler"),
+    )
+    parser.add_argument(
+        "--instrument-open",
+        action="store_true",
+        default=False,
+        help=debug_help("Use pyinstrument web profiler"),
     )
     parser.add_argument(
         "--show-stack",
         "--ss",
         action="store_true",
         default=False,
-        help=SUPPRESS,
+        help=debug_help(
+            r"Use Ctrl+T on macOS or Ctrl+\ on Linux to print the stack "
+            "frame currently executing. Unavailable on Windows."
+        ),
     )

dvc/api/data.py

@@ -36,21 +36,22 @@ def get_url(path, repo=None, rev=None, remote=None):
     NOTE: This function does not check for the actual existence of the file or
     directory in the remote storage.
     """
-    with Repo.open(repo, rev=rev, subrepos=True, uninitialized=True) as _repo:
-        with _wrap_exceptions(_repo, path):
-            fs_path = _repo.dvcfs.from_os_path(path)
+    from dvc.config import NoRemoteError
+    from dvc_data.index import StorageKeyError
 
-            with reraise(FileNotFoundError, PathMissingError(path, repo)):
-                info = _repo.dvcfs.info(fs_path)
-
-            dvc_info = info.get("dvc_info")
-            if not dvc_info:
-                raise OutputNotFoundError(path, repo)
-
-            dvc_repo = info["repo"]  # pylint: disable=unsubscriptable-object
-            md5 = dvc_info["md5"]
-
-            return dvc_repo.cloud.get_url_for(remote, checksum=md5)
+    repo_kwargs: Dict[str, Any] = {}
+    if remote:
+        repo_kwargs["config"] = {"core": {"remote": remote}}
+    with Repo.open(
+        repo, rev=rev, subrepos=True, uninitialized=True, **repo_kwargs
+    ) as _repo:
+        index, entry = _repo.get_data_index_entry(path)
+        with reraise(
+            (StorageKeyError, ValueError),
+            NoRemoteError(f"no remote specified in {_repo}"),
+        ):
+            remote_fs, remote_path = index.storage_map.get_remote(entry)
+            return remote_fs.unstrip_protocol(remote_path)
 
 
 class _OpenContextManager(GCM):
@@ -66,7 +67,7 @@ class _OpenContextManager(GCM):
         raise AttributeError("dvc.api.open() should be used in a with statement.")
 
 
-def open(  # noqa, pylint: disable=redefined-builtin
+def open(  # noqa: A001, pylint: disable=redefined-builtin
     path: str,
     repo: Optional[str] = None,
     rev: Optional[str] = None,
@@ -74,6 +75,7 @@ def open(  # noqa, pylint: disable=redefined-builtin
     mode: str = "r",
     encoding: Optional[str] = None,
     config: Optional[Dict[str, Any]] = None,
+    remote_config: Optional[Dict[str, Any]] = None,
 ):
     """
     Opens a file tracked in a DVC project.
@@ -117,6 +119,9 @@ def open(  # noqa, pylint: disable=redefined-builtin
             Mirrors the namesake parameter in builtin `open()`_.
         config(dict, optional): config to be passed to the DVC repository.
             Defaults to None.
+        remote_config(dict, optional): remote config to be passed to the DVC
+            repository.
+            Defaults to None.
 
     Returns:
         _OpenContextManager: A context manager that generatse a corresponding
@@ -213,22 +218,27 @@ def open(  # noqa, pylint: disable=redefined-builtin
         "mode": mode,
         "encoding": encoding,
         "config": config,
+        "remote_config": remote_config,
     }
     return _OpenContextManager(_open, args, kwargs)
 
 
-def _open(path, repo=None, rev=None, remote=None, mode="r", encoding=None, config=None):
-    if remote:
-        if config is not None:
-            raise ValueError(
-                "can't specify both `remote` and `config` at the same time"
-            )
-        config = {"core": {"remote": remote}}
-
+def _open(
+    path,
+    repo=None,
+    rev=None,
+    remote=None,
+    mode="r",
+    encoding=None,
+    config=None,
+    remote_config=None,
+):
     repo_kwargs: Dict[str, Any] = {
         "subrepos": True,
         "uninitialized": True,
+        "remote": remote,
         "config": config,
+        "remote_config": remote_config,
     }
 
     with Repo.open(repo, rev=rev, **repo_kwargs) as _repo:
@@ -264,7 +274,16 @@ def _open(path, repo=None, rev=None, remote=None, mode="r", encoding=None, confi
                 raise DvcIsADirectoryError(f"'{path}' is a directory") from exc
 
 
-def read(path, repo=None, rev=None, remote=None, mode="r", encoding=None, config=None):
+def read(
+    path,
+    repo=None,
+    rev=None,
+    remote=None,
+    mode="r",
+    encoding=None,
+    config=None,
+    remote_config=None,
+):
     """
     Returns the contents of a tracked file (by DVC or Git). For Git repos, HEAD
     is used unless a rev argument is supplied. The default remote is tried
@@ -278,5 +297,6 @@ def read(path, repo=None, rev=None, remote=None, mode="r", encoding=None, config
         mode=mode,
         encoding=encoding,
         config=config,
+        remote_config=remote_config,
     ) as fd:
         return fd.read()

dvc/cachemgr.py

@@ -24,7 +24,7 @@ def _get_odb(
     if not settings:
         return None
 
-    cls, config, fs_path = get_cloud_fs(repo, **settings)
+    cls, config, fs_path = get_cloud_fs(repo.config, **settings)
     fs = fs or cls(**config)
     if prefix:
         fs_path = fs.path.join(fs_path, *prefix)

dvc/cli/__init__.py

@@ -192,7 +192,7 @@ def main(argv=None):  # noqa: C901, PLR0912, PLR0915
 
             from dvc import PKG, __version__
 
-            pyv = " ".join([python_implementation(), python_version()])
+            pyv = f"{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))
@@ -230,7 +230,7 @@ def main(argv=None):  # noqa: C901, PLR0912, PLR0915
         logger.exception("")
     except DvcParserError:
         ret = 254
-    except Exception as exc:  # noqa, pylint: disable=broad-except
+    except Exception as exc:  # noqa: BLE001, pylint: disable=broad-except
         # pylint: disable=no-member
         ret = _log_exceptions(exc) or 255
 

dvc/cli/parser.py

@@ -137,7 +137,6 @@ def get_parent_parser():
     from dvc._debug import add_debugging_flags
 
     parent_parser = argparse.ArgumentParser(add_help=False)
-    add_debugging_flags(parent_parser)
     log_level_group = parent_parser.add_mutually_exclusive_group()
     log_level_group.add_argument(
         "-q", "--quiet", action="count", default=0, help="Be quiet."
@@ -145,6 +144,7 @@ def get_parent_parser():
     log_level_group.add_argument(
         "-v", "--verbose", action="count", default=0, help="Be verbose."
     )
+    add_debugging_flags(parent_parser)
 
     return parent_parser
 

dvc/cli/utils.py

@@ -1,3 +1,27 @@
+import argparse
+
+
+class DictAction(argparse.Action):
+    def __call__(self, parser, args, values, option_string=None):  # noqa: ARG002
+        d = getattr(args, self.dest) or {}
+
+        if isinstance(values, list):
+            kvs = values
+        else:
+            kvs = [values]
+
+        for kv in kvs:
+            key, value = kv.split("=")
+            if not value:
+                raise argparse.ArgumentError(
+                    self,
+                    f'Could not parse argument "{values}" as k1=v1 k2=v2 ... format',
+                )
+            d[key] = value
+
+        setattr(args, self.dest, d)
+
+
 def fix_subparsers(subparsers):
     """Workaround for bug in Python 3. See more info at:
     https://bugs.python.org/issue16308
@@ -16,9 +40,7 @@ def append_doc_link(help_message, path):
     if not path:
         return help_message
     doc_base = "https://man.dvc.org/"
-    return "{message}\nDocumentation: {link}".format(
-        message=help_message, link=format_link(doc_base + path)
-    )
+    return f"{help_message}\nDocumentation: {format_link(doc_base + path)}"
 
 
 def hide_subparsers_from_help(subparsers):

dvc/commands/commit.py

@@ -47,7 +47,10 @@ def add_parser(subparsers, parent_parser):
         "--force",
         action="store_true",
         default=False,
-        help="Commit even if hash value for dependencies/outputs changed.",
+        help=(
+            "Commit data even if hash values for dependencies or",
+            "outputs did not change.",
+        ),
     )
     commit_parser.add_argument(
         "-d",

dvc/commands/data.py

@@ -90,9 +90,7 @@ class CmdDataStatus(CmdBase):
                     ui.write(f"  ({hint})")
 
             if isinstance(stage_status, dict):
-                items = [
-                    ": ".join([state, file]) for file, state in stage_status.items()
-                ]
+                items = [f"{state}: {file}" for file, state in stage_status.items()]
             else:
                 items = stage_status
 

dvc/commands/data_sync.py

@@ -91,6 +91,8 @@ class CmdDataFetch(CmdDataBase):
                 with_deps=self.args.with_deps,
                 recursive=self.args.recursive,
                 run_cache=self.args.run_cache,
+                max_size=self.args.max_size,
+                types=self.args.types,
             )
             self.log_summary({"fetched": processed_files_count})
         except DvcException:
@@ -322,6 +324,22 @@ def add_parser(subparsers, _parent_parser):
         default=False,
         help="Fetch run history for all stages.",
     )
+    fetch_parser.add_argument(
+        "--max-size",
+        type=int,
+        help="Fetch data files/directories that are each below specified size (bytes).",
+    )
+    fetch_parser.add_argument(
+        "--type",
+        dest="types",
+        action="append",
+        default=[],
+        help=(
+            "Only fetch data files/directories that are of a particular "
+            "type (metrics, plots)."
+        ),
+        choices=["metrics", "plots"],
+    )
     fetch_parser.set_defaults(func=CmdDataFetch)
 
     # Status

dvc/commands/destroy.py

@@ -21,7 +21,7 @@ class CmdDestroy(CmdBase):
             )
 
             if not self.args.force and not ui.confirm(statement):
-                raise DvcException(
+                raise DvcException(  # noqa: TRY301
                     "cannot destroy without a confirmation from the user."
                     " Use `-f` to force."
                 )

dvc/commands/get.py

@@ -3,7 +3,7 @@ import logging
 
 from dvc.cli import completion
 from dvc.cli.command import CmdBaseNoRepo
-from dvc.cli.utils import append_doc_link
+from dvc.cli.utils import DictAction, append_doc_link
 from dvc.exceptions import DvcException
 
 logger = logging.getLogger(__name__)
@@ -37,6 +37,9 @@ class CmdGet(CmdBaseNoRepo):
                 rev=self.args.rev,
                 jobs=self.args.jobs,
                 force=self.args.force,
+                config=self.args.config,
+                remote=self.args.remote,
+                remote_config=self.args.remote_config,
             )
             return 0
         except CloneError:
@@ -102,4 +105,27 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Override local file or folder if exists.",
     )
+    get_parser.add_argument(
+        "--config",
+        type=str,
+        help=(
+            "Path to a config file that will be merged with the config "
+            "in the target repository."
+        ),
+    )
+    get_parser.add_argument(
+        "--remote",
+        type=str,
+        help="Remote name to set as a default in the target repository.",
+    )
+    get_parser.add_argument(
+        "--remote-config",
+        type=str,
+        nargs="*",
+        action=DictAction,
+        help=(
+            "Remote config options to merge with a remote's config (default or one "
+            "specified by '--remote') in the target repository."
+        ),
+    )
     get_parser.set_defaults(func=CmdGet)

dvc/commands/imp.py

@@ -3,7 +3,7 @@ import logging
 
 from dvc.cli import completion
 from dvc.cli.command import CmdBase
-from dvc.cli.utils import append_doc_link
+from dvc.cli.utils import DictAction, append_doc_link
 from dvc.exceptions import DvcException
 
 logger = logging.getLogger(__name__)
@@ -22,6 +22,9 @@ class CmdImport(CmdBase):
                 no_exec=self.args.no_exec,
                 no_download=self.args.no_download,
                 jobs=self.args.jobs,
+                config=self.args.config,
+                remote=self.args.remote,
+                remote_config=self.args.remote_config,
             )
         except CloneError:
             logger.exception("failed to import '%s'", self.args.path)
@@ -94,4 +97,27 @@ def add_parser(subparsers, parent_parser):
         ),
         metavar="<number>",
     )
+    import_parser.add_argument(
+        "--config",
+        type=str,
+        help=(
+            "Path to a config file that will be merged with the config "
+            "in the target repository."
+        ),
+    )
+    import_parser.add_argument(
+        "--remote",
+        type=str,
+        help="Remote name to set as a default in the target repository.",
+    )
+    import_parser.add_argument(
+        "--remote-config",
+        type=str,
+        nargs="*",
+        action=DictAction,
+        help=(
+            "Remote config options to merge with a remote's config (default or one "
+            "specified by '--remote') in the target repository."
+        ),
+    )
     import_parser.set_defaults(func=CmdImport)

dvc/commands/ls/__init__.py

@@ -3,7 +3,7 @@ import logging
 
 from dvc.cli import completion
 from dvc.cli.command import CmdBaseNoRepo
-from dvc.cli.utils import append_doc_link
+from dvc.cli.utils import DictAction, append_doc_link
 from dvc.commands.ls.ls_colors import LsColors
 from dvc.exceptions import DvcException
 from dvc.ui import ui
@@ -34,6 +34,9 @@ class CmdList(CmdBaseNoRepo):
                 rev=self.args.rev,
                 recursive=self.args.recursive,
                 dvc_only=self.args.dvc_only,
+                config=self.args.config,
+                remote=self.args.remote,
+                remote_config=self.args.remote_config,
             )
             if self.args.json:
                 ui.write_json(entries)
@@ -80,6 +83,29 @@ def add_parser(subparsers, parent_parser):
         help="Git revision (e.g. SHA, branch, tag)",
         metavar="<commit>",
     )
+    list_parser.add_argument(
+        "--config",
+        type=str,
+        help=(
+            "Path to a config file that will be merged with the config "
+            "in the target repository."
+        ),
+    )
+    list_parser.add_argument(
+        "--remote",
+        type=str,
+        help="Remote name to set as a default in the target repository.",
+    )
+    list_parser.add_argument(
+        "--remote-config",
+        type=str,
+        nargs="*",
+        action=DictAction,
+        help=(
+            "Remote config options to merge with a remote's config (default or one "
+            "specified by '--remote') in the target repository."
+        ),
+    )
     list_parser.add_argument(
         "path",
         nargs="?",

dvc/commands/remote.py

@@ -127,9 +127,7 @@ class CmdRemoteRename(CmdRemote):
         all_config = self.config.load_config_to_level(None)
         if self.args.new in all_config.get("remote", {}):
             raise ConfigError(
-                "Rename failed. Remote name '{}' already exists.".format(
-                    {self.args.new}
-                )
+                f"Rename failed. Remote name {self.args.new!r} already exists."
             )
 
         with self.config.edit(self.args.level) as conf:

dvc/commands/repro.py

@@ -33,6 +33,7 @@ class CmdRepro(CmdBase):
             "force_downstream": self.args.force_downstream,
             "pull": self.args.pull,
             "allow_missing": self.args.allow_missing,
+            "on_error": self.args.on_error,
         }
 
     @property
@@ -137,6 +138,26 @@ and then the stage name name.
             "Only print the commands that would be executed without actually executing."
         ),
     )
+    repro_parser.add_argument(
+        "-k",
+        "--keep-going",
+        action="store_const",
+        default="fail",
+        const="keep-going",
+        dest="on_error",
+        help=(
+            "Continue executing, skipping stages having dependencies "
+            "on the failed stages"
+        ),
+    )
+    repro_parser.add_argument(
+        "--ignore-errors",
+        action="store_const",
+        default="fail",
+        const="ignore",
+        dest="on_error",
+        help="Ignore errors from stages.",
+    )
 
 
 def add_parser(subparsers, parent_parser):

dvc/commands/stage.py

@@ -195,12 +195,6 @@ def _add_common_args(parser):
         help="Declare output file or directory (do not put into DVC cache).",
         metavar="<filename>",
     ).complete = completion.FILE
-    parser.add_argument(
-        "--external",
-        action="store_true",
-        default=False,
-        help="Allow outputs that are outside of the DVC repository.",
-    )
     parser.add_argument(
         "--outs-persist",
         action="append",

dvc/compare.py

@@ -57,7 +57,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
         return self._columns[name]
 
     def items(self) -> ItemsView[str, Column]:
-        projection = {k: self.column(k) for k in self.keys()}  # noqa: SIM118
+        projection = {k: self.column(k) for k in self.keys()}
         return projection.items()
 
     def keys(self) -> List[str]:
@@ -171,8 +171,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
                 self.add_column(key)
 
         row: List["CellT"] = [
-            with_value(d.get(key), self._fill_value)
-            for key in self.keys()  # noqa: SIM118
+            with_value(d.get(key), self._fill_value) for key in self.keys()
         ]
         self.append(row)
 
@@ -227,7 +226,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
             to_drop -= match_line
 
         if axis == "rows":
-            for name in self.keys():  # noqa: SIM118
+            for name in self.keys():
                 self._columns[name] = Column(
                     [x for n, x in enumerate(self._columns[name]) if n not in to_drop]
                 )
@@ -275,7 +274,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
                 else:
                     unique_rows.append(tuple_row)
 
-            for name in self.keys():  # noqa: SIM118
+            for name in self.keys():
                 self._columns[name] = Column(
                     [
                         x

dvc/config.py

@@ -93,6 +93,8 @@ class Config(dict):
         validate: bool = True,
         fs: Optional["FileSystem"] = None,
         config: Optional["DictStrAny"] = None,
+        remote: Optional[str] = None,
+        remote_config: Optional["DictStrAny"] = None,
     ):  # pylint: disable=super-init-not-called
         from dvc.fs import LocalFileSystem
 
@@ -101,9 +103,11 @@ class Config(dict):
         self.fs = fs or self.wfs
 
         if dvc_dir:
-            self.dvc_dir = self.fs.path.abspath(self.fs.path.realpath(dvc_dir))
+            self.dvc_dir = self.fs.path.abspath(dvc_dir)
 
-        self.load(validate=validate, config=config)
+        self.load(
+            validate=validate, config=config, remote=remote, remote_config=remote_config
+        )
 
     @classmethod
     def from_cwd(cls, fs: Optional["FileSystem"] = None, **kwargs):
@@ -154,7 +158,13 @@ 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,
+        remote: Optional[str] = None,
+        remote_config: Optional["DictStrAny"] = None,
+    ):
         """Loads config from all the config files.
 
         Raises:
@@ -169,6 +179,17 @@ class Config(dict):
             conf = self.validate(conf)
 
         self.clear()
+
+        if remote:
+            conf["core"]["remote"] = remote
+
+        if remote_config:
+            remote = remote or conf["core"].get("remote")
+            if not remote:
+                raise ValueError("Missing remote name")
+
+            merge(conf, {"remote": {remote: remote_config}})
+
         self.update(conf)
 
     def _get_fs(self, level):
@@ -176,23 +197,32 @@ class Config(dict):
         # the repo.
         return self.fs if level == "repo" else self.wfs
 
-    def _load_config(self, level):
+    @staticmethod
+    def load_file(path, fs=None) -> dict:
         from configobj import ConfigObj, ConfigObjError
 
+        from dvc.fs import localfs
+
+        fs = fs or localfs
+
+        with fs.open(path) as fobj:
+            try:
+                conf_obj = ConfigObj(fobj)
+            except UnicodeDecodeError as exc:
+                raise ConfigError(str(exc)) from exc
+            except ConfigObjError as exc:
+                raise ConfigError(str(exc)) from exc
+
+        return _parse_named(_lower_keys(conf_obj.dict()))
+
+    def _load_config(self, level):
         filename = self.files[level]
         fs = self._get_fs(level)
 
-        if fs.exists(filename):
-            with fs.open(filename) as fobj:
-                try:
-                    conf_obj = ConfigObj(fobj)
-                except UnicodeDecodeError as exc:
-                    raise ConfigError(str(exc)) from exc
-                except ConfigObjError as exc:
-                    raise ConfigError(str(exc)) from exc
-        else:
-            conf_obj = ConfigObj()
-        return _parse_named(_lower_keys(conf_obj.dict()))
+        try:
+            return self.load_file(filename, fs=fs)
+        except FileNotFoundError:
+            return {}
 
     def _save_config(self, level, conf_dict):
         from configobj import ConfigObj

dvc/config_schema.py

@@ -75,7 +75,7 @@ def ByUrl(mapping):  # noqa: N802
 
 
 class RelPath(str):
-    pass
+    __slots__ = ()
 
 
 class FeatureSchema(Schema):

dvc/daemon.py

@@ -28,7 +28,9 @@ def _popen(cmd, **kwargs) -> Popen:
             os.path.abspath(os.path.dirname(__file__)), "__main__.py"
         )
         prefix += [main_entrypoint]
-    return Popen(prefix + cmd, close_fds=True, shell=False, **kwargs)  # nosec B603
+    return Popen(
+        prefix + cmd, close_fds=True, shell=False, **kwargs  # nosec B603  # noqa: S603
+    )
 
 
 def _spawn_windows(cmd, env):

dvc/dagascii.py

@@ -209,6 +209,30 @@ def draw(vertices, edges):
     Args:
         vertices (list): list of graph vertices.
         edges (list): list of graph edges.
+
+    Returns:
+        str: ASCII representation
+
+    Example:
+        >>> from dvc.dagascii import draw
+        >>> vertices = [1, 2, 3, 4]
+        >>> edges = [(1, 2), (2, 3), (2, 4), (1, 4)]
+        >>> print(draw(vertices, edges))
+        +---+     +---+
+        | 3 |     | 4 |
+        +---+    *+---+
+          *    **   *
+          *  **     *
+          * *       *
+        +---+       *
+        | 2 |      *
+        +---+     *
+             *    *
+              *  *
+               **
+             +---+
+             | 1 |
+             +---+
     """
     # pylint: disable=too-many-locals
     # NOTE: coordinates might me negative, so we need to shift

dvc/data_cloud.py

@@ -4,6 +4,7 @@ import logging
 from typing import TYPE_CHECKING, Iterable, Optional, Set, Tuple
 
 from dvc.config import NoRemoteError, RemoteConfigError
+from dvc.fs.callbacks import Callback
 from dvc.utils.objects import cached_property
 from dvc_data.hashfile.db import get_index
 from dvc_data.hashfile.transfer import TransferResult
@@ -90,7 +91,7 @@ class DataCloud:
         if name:
             from dvc.fs import get_cloud_fs
 
-            cls, config, fs_path = get_cloud_fs(self.repo, name=name)
+            cls, config, fs_path = get_cloud_fs(self.repo.config, name=name)
 
             if config.get("worktree"):
                 version_aware = config.get("version_aware")
@@ -209,15 +210,20 @@ class DataCloud:
             cache = self.repo.cache.legacy
         else:
             cache = self.repo.cache.local
-        return self.transfer(
-            cache,
-            odb,
-            objs,
-            jobs=jobs,
-            dest_index=get_index(odb),
-            cache_odb=cache,
-            validate_status=self._log_missing,
-        )
+        with Callback.as_tqdm_callback(
+            desc=f"Pushing to {odb.fs.unstrip_protocol(odb.path)}",
+            unit="file",
+        ) as cb:
+            return self.transfer(
+                cache,
+                odb,
+                objs,
+                jobs=jobs,
+                dest_index=get_index(odb),
+                cache_odb=cache,
+                validate_status=self._log_missing,
+                callback=cb,
+            )
 
     def pull(
         self,
@@ -263,16 +269,21 @@ class DataCloud:
             cache = self.repo.cache.legacy
         else:
             cache = self.repo.cache.local
-        return self.transfer(
-            odb,
-            cache,
-            objs,
-            jobs=jobs,
-            src_index=get_index(odb),
-            cache_odb=cache,
-            verify=odb.verify,
-            validate_status=self._log_missing,
-        )
+        with Callback.as_tqdm_callback(
+            desc=f"Fetching from {odb.fs.unstrip_protocol(odb.path)}",
+            unit="file",
+        ) as cb:
+            return self.transfer(
+                odb,
+                cache,
+                objs,
+                jobs=jobs,
+                src_index=get_index(odb),
+                cache_odb=cache,
+                verify=odb.verify,
+                validate_status=self._log_missing,
+                callback=cb,
+            )
 
     def status(
         self,

dvc/dependency/base.py

@@ -50,7 +50,7 @@ class Dependency(Output):
             self.fs_path = self.fs.path.version_path(self.fs_path, self.meta.version_id)
 
     def download(self, to, jobs=None):
-        fs_download(self.fs, self.fs_path, to, jobs=jobs)
+        fs_download(self.fs, self.fs_path, to.fs_path, jobs=jobs)
 
     def save(self):
         super().save()

dvc/dependency/repo.py

@@ -1,6 +1,6 @@
 from typing import TYPE_CHECKING, Dict, Optional, Union
 
-from voluptuous import Required
+from voluptuous import Any, Required
 
 from dvc.utils import as_posix
 
@@ -16,12 +16,16 @@ class RepoDependency(Dependency):
     PARAM_URL = "url"
     PARAM_REV = "rev"
     PARAM_REV_LOCK = "rev_lock"
+    PARAM_CONFIG = "config"
+    PARAM_REMOTE = "remote"
 
     REPO_SCHEMA = {
         PARAM_REPO: {
             Required(PARAM_URL): str,
             PARAM_REV: str,
             PARAM_REV_LOCK: str,
+            PARAM_CONFIG: str,
+            PARAM_REMOTE: Any(str, dict),
         }
     }
 
@@ -60,7 +64,28 @@ class RepoDependency(Dependency):
             self.def_repo[self.PARAM_REV_LOCK] = rev
 
     def dumpd(self, **kwargs) -> Dict[str, Union[str, Dict[str, str]]]:
-        return {self.PARAM_PATH: self.def_path, self.PARAM_REPO: self.def_repo}
+        repo = {self.PARAM_URL: self.def_repo[self.PARAM_URL]}
+
+        rev = self.def_repo.get(self.PARAM_REV)
+        if rev:
+            repo[self.PARAM_REV] = self.def_repo[self.PARAM_REV]
+
+        rev_lock = self.def_repo.get(self.PARAM_REV_LOCK)
+        if rev_lock:
+            repo[self.PARAM_REV_LOCK] = rev_lock
+
+        config = self.def_repo.get(self.PARAM_CONFIG)
+        if config:
+            repo[self.PARAM_CONFIG] = config
+
+        remote = self.def_repo.get(self.PARAM_REMOTE)
+        if remote:
+            repo[self.PARAM_REMOTE] = remote
+
+        return {
+            self.PARAM_PATH: self.def_path,
+            self.PARAM_REPO: repo,
+        }
 
     def update(self, rev: Optional[str] = None):
         if rev:
@@ -77,9 +102,24 @@ class RepoDependency(Dependency):
     def _make_fs(
         self, rev: Optional[str] = None, locked: bool = True
     ) -> "DVCFileSystem":
+        from dvc.config import Config
         from dvc.fs import DVCFileSystem
 
-        config = {"cache": self.repo.config["cache"]}
+        rem = self.def_repo.get("remote")
+        if isinstance(rem, dict):
+            remote = None  # type: ignore[unreachable]
+            remote_config = rem
+        else:
+            remote = rem
+            remote_config = None
+
+        conf = self.def_repo.get("config")
+        if conf:
+            config = Config.load_file(conf)
+        else:
+            config = {}
+
+        config["cache"] = self.repo.config["cache"]
         config["cache"]["dir"] = self.repo.cache.local_cache_dir
 
         return DVCFileSystem(
@@ -87,6 +127,8 @@ class RepoDependency(Dependency):
             rev=rev or self._get_rev(locked=locked),
             subrepos=True,
             config=config,
+            remote=remote,
+            remote_config=remote_config,
         )
 
     def _get_rev(self, locked: bool = True):

dvc/dvcfile.py

@@ -97,9 +97,7 @@ class FileMixin:
         self.verify = verify
 
     def __repr__(self):
-        return "{}: {}".format(
-            self.__class__.__name__, relpath(self.path, self.repo.root_dir)
-        )
+        return f"{self.__class__.__name__}: {relpath(self.path, self.repo.root_dir)}"
 
     def __hash__(self):
         return hash(self.path)

dvc/exceptions.py

@@ -17,6 +17,10 @@ class DvcException(Exception):
 class InvalidArgumentError(ValueError, DvcException):
     """Thrown if arguments are invalid."""
 
+    def __init__(self, msg, *args):
+        self.msg = msg
+        super().__init__(msg, *args)
+
 
 class OutputDuplicationError(DvcException):
     """Thrown if a file/directory is specified as an output in more than one
@@ -63,12 +67,14 @@ class OutputNotFoundError(DvcException):
         self.output = output
         self.repo = repo
         super().__init__(
-            "Unable to find DVC file with output '{path}'".format(
-                path=relpath(self.output)
-            )
+            f"Unable to find DVC file with output {relpath(self.output)!r}"
         )
 
 
+class StageNotFoundError(DvcException):
+    pass
+
+
 class StagePathAsOutputError(DvcException):
     """Thrown if directory that stage is going to be saved in is specified as
     an output of another stage.
@@ -80,11 +86,7 @@ class StagePathAsOutputError(DvcException):
 
     def __init__(self, stage, output):
         assert isinstance(output, str)
-        super().__init__(
-            "{stage} is within an output '{output}' of another stage".format(
-                stage=stage, output=output
-            )
-        )
+        super().__init__(f"{stage} is within an output {output!r} of another stage")
 
 
 class CircularDependencyError(DvcException):
@@ -156,9 +158,7 @@ class CyclicGraphError(DvcException):
 class ConfirmRemoveError(DvcException):
     def __init__(self, path):
         super().__init__(
-            "unable to remove '{}' without a confirmation. Use `-f` to force.".format(
-                path
-            )
+            f"unable to remove {path!r} without a confirmation. Use `-f` to force."
         )
 
 
@@ -167,9 +167,7 @@ class InitError(DvcException):
 
 
 class ReproductionError(DvcException):
-    def __init__(self, name):
-        self.name = name
-        super().__init__(f"failed to reproduce '{name}'")
+    pass
 
 
 class BadMetricError(DvcException):
@@ -299,7 +297,7 @@ class URLMissingError(DvcException):
         super().__init__(f"The path '{url}' does not exist")
 
 
-class IsADirectoryError(DvcException):  # noqa,pylint:disable=redefined-builtin
+class IsADirectoryError(DvcException):  # noqa: A001,pylint:disable=redefined-builtin
     """Raised when a file operation is requested on a directory."""
 
 

dvc/fs/__init__.py

@@ -1,3 +1,4 @@
+from typing import Optional
 from urllib.parse import urlparse
 
 from dvc_http import HTTPFileSystem, HTTPSFileSystem  # noqa: F401
@@ -18,7 +19,7 @@ from dvc_objects.fs import (  # noqa: F401
     system,
     utils,
 )
-from dvc_objects.fs.base import AnyFSPath, FileSystem  # noqa: F401
+from dvc_objects.fs.base import AnyFSPath, FileSystem  # noqa: F401, TCH001
 from dvc_objects.fs.errors import (  # noqa: F401
     AuthError,
     ConfigError,
@@ -48,18 +49,45 @@ known_implementations.update(
 # pylint: enable=unused-import
 
 
-def download(fs, fs_path, to, jobs=None):
+def download(fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None):
     with Callback.as_tqdm_callback(
         desc=f"Downloading {fs.path.name(fs_path)}",
         unit="files",
     ) as cb:
-        fs.get(fs_path, to.fs_path, batch_size=jobs, callback=cb)
+        # NOTE: We use dvc-objects generic.copy over fs.get since it makes file
+        # download atomic and avoids fsspec glob/regex path expansion.
+        if fs.isdir(fs_path):
+            from_infos = [
+                path
+                for path in fs.find(fs_path)
+                if not path.endswith(fs.path.flavour.sep)
+            ]
+            if not from_infos:
+                return localfs.makedirs(to, exist_ok=True)
+            to_infos = [
+                localfs.path.join(to, *fs.path.relparts(info, fs_path))
+                for info in from_infos
+            ]
+        else:
+            from_infos = [fs_path]
+            to_infos = [to]
+
+        cb.set_size(len(from_infos))
+        jobs = jobs or fs.jobs
+        generic.copy(
+            fs,
+            from_infos,
+            localfs,
+            to_infos,
+            callback=cb,
+            batch_size=jobs,
+        )
 
 
 def parse_external_url(url, config=None):
     remote_config = dict(config) if config else {}
     remote_config["url"] = url
-    fs_cls, fs_config, fs_path = get_cloud_fs(None, **remote_config)
+    fs_cls, fs_config, fs_path = get_cloud_fs({}, **remote_config)
     fs = fs_cls(**fs_config)
     return fs, fs_path
 
@@ -105,18 +133,14 @@ def _resolve_remote_refs(config, remote_conf):
         return remote_conf
 
     base = get_fs_config(config, name=parsed.netloc)
-    cls, _, _ = _get_cloud_fs(config, **base)
+    cls, _, _ = get_cloud_fs(config, **base)
     relpath = parsed.path.lstrip("/").replace("/", cls.sep)
     url = cls.sep.join((base["url"], relpath))
     return {**base, **remote_conf, "url": url}
 
 
-def get_cloud_fs(repo, **kwargs):
-    repo_config = repo.config if repo else {}
-    return _get_cloud_fs(repo_config, **kwargs)
-
-
-def _get_cloud_fs(repo_config, **kwargs):
+def get_cloud_fs(repo_config, **kwargs):
+    repo_config = repo_config or {}
     core_config = repo_config.get("core", {})
 
     remote_conf = get_fs_config(repo_config, **kwargs)

dvc/fs/dvc.py

@@ -5,7 +5,7 @@ import ntpath
 import os
 import posixpath
 import threading
-from contextlib import suppress
+from contextlib import ExitStack, suppress
 from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Tuple, Type, Union
 
 from fsspec.spec import AbstractFileSystem
@@ -54,6 +54,8 @@ def _merge_info(repo, key, fs_info, dvc_info):
         ret["size"] = dvc_info["size"]
         if not fs_info and "md5" in dvc_info:
             ret["md5"] = dvc_info["md5"]
+        if not fs_info and "md5-dos2unix" in dvc_info:
+            ret["md5-dos2unix"] = dvc_info["md5-dos2unix"]
 
     if fs_info:
         ret["type"] = fs_info["type"]
@@ -121,11 +123,13 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         from pygtrie import Trie
 
         super().__init__()
+        self._repo_stack = ExitStack()
         if repo is None:
             repo = self._make_repo(url=url, rev=rev, subrepos=subrepos, **repo_kwargs)
             assert repo is not None
             # pylint: disable=protected-access
             repo_factory = repo._fs_conf["repo_factory"]
+            self._repo_stack.enter_context(repo)
 
         if not repo_factory:
             from dvc.repo import Repo
@@ -214,6 +218,7 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
                     scm=self.repo.scm,
                     repo_factory=self.repo_factory,
                 )
+                self._repo_stack.enter_context(repo)
                 self._datafss[key] = DataFileSystem(index=repo.index.data["repo"])
             self._subrepos_trie[key] = repo
 
@@ -283,7 +288,7 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         dvc_infos = {}
         if dvc_fs:
             dvc_path = _get_dvc_path(dvc_fs, subkey)
-            with suppress(FileNotFoundError):
+            with suppress(FileNotFoundError, NotADirectoryError):
                 for info in dvc_fs.ls(dvc_path, detail=True):
                     dvc_infos[dvc_fs.path.name(info["name"])] = info
             dvc_exists = bool(dvc_infos) or dvc_fs.exists(dvc_path)
@@ -347,7 +352,7 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
                 dvc_info = dvc_fs.fs.index.info(subkey)
                 dvc_path = _get_dvc_path(dvc_fs, subkey)
                 dvc_info["name"] = dvc_path
-            except FileNotFoundError:
+            except KeyError:
                 pass
 
         fs_info = None
@@ -394,6 +399,9 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         dvc_path = _get_dvc_path(dvc_fs, subkey)
         return dvc_fs.get_file(dvc_path, lpath, **kwargs)
 
+    def close(self):
+        self._repo_stack.close()
+
 
 class DVCFileSystem(FileSystem):
     protocol = "local"
@@ -404,7 +412,7 @@ class DVCFileSystem(FileSystem):
 
     @functools.cached_property
     # pylint: disable-next=invalid-overridden-method
-    def fs(self) -> "DVCFileSystem":
+    def fs(self) -> "_DVCFileSystem":
         return _DVCFileSystem(**self.fs_args)
 
     def isdvc(self, path, **kwargs) -> bool:
@@ -427,3 +435,7 @@ class DVCFileSystem(FileSystem):
             path = os.path.relpath(path, self.repo.root_dir)
 
         return as_posix(path)
+
+    def close(self):
+        if "fs" in self.__dict__:
+            self.fs.close()

dvc/logger.py

@@ -56,7 +56,7 @@ def add_logging_level(level_name, level_num, method_name=None):
 
 class LoggingException(Exception):
     def __init__(self, record):
-        msg = f"failed to log {str(record)}"
+        msg = f"failed to log {record!s}"
         super().__init__(msg)
 
 
@@ -160,7 +160,7 @@ class LoggerHandler(logging.StreamHandler):
             self.flush()
         except (BrokenPipeError, RecursionError):
             raise
-        except Exception:  # noqa, pylint: disable=broad-except
+        except Exception:  # noqa: BLE001, pylint: disable=broad-except
             self.handleError(record)
 
 

dvc/output.py

@@ -354,7 +354,11 @@ class Output:
         if fs_config is not None:
             fs_kwargs.update(**fs_config)
 
-        fs_cls, fs_config, fs_path = get_cloud_fs(self.repo, url=path, **fs_kwargs)
+        fs_cls, fs_config, fs_path = get_cloud_fs(
+            self.repo.config if self.repo else {},
+            url=path,
+            **fs_kwargs,
+        )
         self.fs = fs_cls(**fs_config)
 
         if (
@@ -471,9 +475,7 @@ class Output:
         return fs.path.abspath(fs.path.normpath(fs_path))
 
     def __repr__(self):
-        return "{class_name}: '{def_path}'".format(
-            class_name=type(self).__name__, def_path=self.def_path
-        )
+        return f"{type(self).__name__}: {self.def_path!r}"
 
     def __str__(self):
         if self.fs.protocol != "local":
@@ -486,6 +488,9 @@ class Output:
         ):
             return str(self.def_path)
 
+        if not self.fs.path.isin(self.fs_path, self.repo.root_dir):
+            return self.fs_path
+
         cur_dir = self.fs.path.getcwd()
         if self.fs.path.isin(cur_dir, self.repo.root_dir):
             return self.fs.path.relpath(self.fs_path, cur_dir)
@@ -548,12 +553,24 @@ class Output:
         _, hash_info = self._get_hash_meta()
         return hash_info
 
+    def _build(
+        self, *args, no_progress_bar=False, **kwargs
+    ) -> Tuple["HashFileDB", "Meta", "HashFile"]:
+        from dvc.ui import ui
+
+        with ui.progress(
+            unit="file",
+            desc=f"Collecting files and computing hashes in {self}",
+            disable=no_progress_bar,
+        ) as pb:
+            return build(*args, callback=pb.as_callback(), **kwargs)
+
     def _get_hash_meta(self):
         if self.use_cache:
             odb = self.cache
         else:
             odb = self.local_cache
-        _, meta, obj = build(
+        _, meta, obj = self._build(
             odb,
             self.fs_path,
             self.fs,
@@ -681,7 +698,7 @@ class Output:
     def save(self) -> None:
         if self.use_cache and not self.is_in_repo:
             raise DvcException(
-                f"Saving cached external output {str(self)} is not supported "
+                f"Saving cached external output {self!s} is not supported "
                 "since DVC 3.0. See "
                 f"{format_link('https://dvc.org/doc/user-guide/upgrade')} "
                 "for more info."
@@ -703,7 +720,7 @@ class Output:
 
         self._update_legacy_hash_name()
         if self.use_cache:
-            _, self.meta, self.obj = build(
+            _, self.meta, self.obj = self._build(
                 self.cache,
                 self.fs_path,
                 self.fs,
@@ -711,7 +728,7 @@ class Output:
                 ignore=self.dvcignore,
             )
         else:
-            _, self.meta, self.obj = build(
+            _, self.meta, self.obj = self._build(
                 self.local_cache,
                 self.fs_path,
                 self.fs,
@@ -763,20 +780,25 @@ class Output:
             if granular:
                 obj = self._commit_granular_dir(filter_info, hardlink)
             else:
-                staging, _, obj = build(
+                staging, _, obj = self._build(
                     self.cache,
                     filter_info or self.fs_path,
                     self.fs,
                     self.hash_name,
                     ignore=self.dvcignore,
                 )
-                otransfer(
-                    staging,
-                    self.cache,
-                    {obj.hash_info},
-                    shallow=False,
-                    hardlink=hardlink,
-                )
+                with Callback.as_tqdm_callback(
+                    desc=f"Committing {self} to cache",
+                    unit="file",
+                ) as cb:
+                    otransfer(
+                        staging,
+                        self.cache,
+                        {obj.hash_info},
+                        shallow=False,
+                        hardlink=hardlink,
+                        callback=cb,
+                    )
             if relink:
                 rel = self.fs.path.relpath(filter_info or self.fs_path)
                 with CheckoutCallback(desc=f"Checking out {rel}", unit="files") as cb:
@@ -794,7 +816,7 @@ class Output:
 
     def _commit_granular_dir(self, filter_info, hardlink) -> Optional["HashFile"]:
         prefix = self.fs.path.parts(self.fs.path.relpath(filter_info, self.fs_path))
-        staging, _, obj = build(
+        staging, _, obj = self._build(
             self.cache,
             self.fs_path,
             self.fs,
@@ -805,13 +827,18 @@ class Output:
         save_obj = obj.filter(prefix)
         assert isinstance(save_obj, Tree)
         checkout_obj = save_obj.get_obj(self.cache, prefix)
-        otransfer(
-            staging,
-            self.cache,
-            {save_obj.hash_info} | {oid for _, _, oid in save_obj},
-            shallow=True,
-            hardlink=hardlink,
-        )
+        with Callback.as_tqdm_callback(
+            desc=f"Committing {self} to cache",
+            unit="file",
+        ) as cb:
+            otransfer(
+                staging,
+                self.cache,
+                {save_obj.hash_info} | {oid for _, _, oid in save_obj},
+                shallow=True,
+                hardlink=hardlink,
+                callback=cb,
+            )
         return checkout_obj
 
     def dumpd(self, **kwargs):  # noqa: C901, PLR0912
@@ -1007,7 +1034,9 @@ class Output:
         if odb is None:
             odb = self.cache
 
-        cls, config, from_info = get_cloud_fs(self.repo, url=source)
+        cls, config, from_info = get_cloud_fs(
+            self.repo.config if self.repo else {}, url=source
+        )
         from_fs = cls(**config)
 
         # When running import-url --to-remote / add --to-remote/-o ... we
@@ -1021,7 +1050,7 @@ class Output:
 
         upload = not (update and from_fs.isdir(from_info))
         jobs = jobs or min((from_fs.jobs, odb.fs.jobs))
-        staging, self.meta, obj = build(
+        staging, self.meta, obj = self._build(
             odb,
             from_info,
             from_fs,
@@ -1029,14 +1058,19 @@ class Output:
             upload=upload,
             no_progress_bar=no_progress_bar,
         )
-        otransfer(
-            staging,
-            odb,
-            {obj.hash_info},
-            jobs=jobs,
-            hardlink=False,
-            shallow=False,
-        )
+        with Callback.as_tqdm_callback(
+            desc=f"Transferring to {odb.fs.unstrip_protocol(odb.path)}",
+            unit="file",
+        ) as cb:
+            otransfer(
+                staging,
+                odb,
+                {obj.hash_info},
+                jobs=jobs,
+                hardlink=False,
+                shallow=False,
+                callback=cb,
+            )
 
         self.hash_info = obj.hash_info
         self.files = None
@@ -1357,7 +1391,7 @@ class Output:
         new: "HashFile"
         try:
             assert self.hash_name
-            staging, meta, obj = build(
+            staging, meta, obj = self._build(
                 cache,
                 path,
                 self.fs,
@@ -1399,10 +1433,23 @@ class Output:
 
         assert staging
         assert obj.hash_info
-        otransfer(staging, self.cache, {obj.hash_info}, hardlink=relink, shallow=False)
+        with Callback.as_tqdm_callback(
+            desc=f"Adding {self} to cache",
+            unit="file",
+        ) as cb:
+            otransfer(
+                staging,
+                self.cache,
+                {obj.hash_info},
+                hardlink=relink,
+                shallow=False,
+                callback=cb,
+            )
 
         if relink:
-            with CheckoutCallback(desc="Checking out {path}", unit="files") as callback:
+            with CheckoutCallback(
+                desc=f"Checking out {path}", unit="files"
+            ) as callback:
                 self._checkout(
                     path,
                     self.fs,

dvc/parsing/__init__.py

@@ -135,6 +135,7 @@ def make_definition(
 class DataResolver:
     def __init__(self, repo: "Repo", wdir: str, d: dict):
         self.fs = fs = repo.fs
+        self.parsing_config = repo.config.get("parsing", {})
 
         if os.path.isabs(wdir):
             wdir = fs.path.relpath(wdir)
@@ -301,7 +302,10 @@ class EntryDefinition:
     ) -> "DictStrAny":
         try:
             return context.resolve(
-                value, skip_interpolation_checks=skip_checks, key=key
+                value,
+                skip_interpolation_checks=skip_checks,
+                key=key,
+                config=self.resolver.parsing_config,
             )
         except (ParseError, KeyNotInContext) as exc:
             format_and_raise(exc, f"'{self.where}.{self.name}.{key}'", self.relpath)

dvc/parsing/context.py

@@ -495,7 +495,12 @@ class Context(CtxDict):
                 self.data.pop(key, None)
 
     def resolve(
-        self, src, unwrap=True, skip_interpolation_checks=False, key=None
+        self,
+        src,
+        unwrap=True,
+        skip_interpolation_checks=False,
+        key=None,
+        config=None,
     ) -> Any:
         """Recursively resolves interpolation and returns resolved data.
 
@@ -511,10 +516,15 @@ class Context(CtxDict):
         {'lst': [1, 2, 3]}
         """
         func = recurse(self.resolve_str)
-        return func(src, unwrap, skip_interpolation_checks, key)
+        return func(src, unwrap, skip_interpolation_checks, key, config)
 
     def resolve_str(
-        self, src: str, unwrap=True, skip_interpolation_checks=False, key=None
+        self,
+        src: str,
+        unwrap=True,
+        skip_interpolation_checks=False,
+        key=None,
+        config=None,
     ) -> str:
         """Resolves interpolated string to it's original value,
         or in case of multiple interpolations, a combined string.
@@ -535,7 +545,12 @@ class Context(CtxDict):
             return value
         # but not "${num} days"
         return str_interpolate(
-            src, matches, self, skip_checks=skip_interpolation_checks, key=key
+            src,
+            matches,
+            self,
+            skip_checks=skip_interpolation_checks,
+            key=key,
+            config=config,
         )
 
 

dvc/parsing/interpolate.py

@@ -82,20 +82,18 @@ def escape_str(value):
 
 
 @singledispatch
-def to_str(obj) -> str:
+def to_str(obj, config=None) -> str:  # noqa: ARG001, pylint: disable=unused-argument
     return str(obj)
 
 
 @to_str.register(bool)
-def _(obj: bool):
+def _(obj: bool, config=None):  # noqa: ARG001, pylint: disable=unused-argument
     return "true" if obj else "false"
 
 
 @to_str.register(dict)
-def _(obj: dict):  # noqa: C901
-    from dvc.config import Config
-
-    config = Config.from_cwd().get("parsing", {})
+def _(obj: dict, config=None):  # noqa: C901
+    config = config or {}
 
     result = ""
     for k, v in flatten(obj).items():
@@ -211,6 +209,7 @@ def str_interpolate(
     context: "Context",
     skip_checks: bool = False,
     key=None,
+    config=None,
 ):
     index, buf = 0, ""
     for match in matches:
@@ -218,7 +217,7 @@ def str_interpolate(
         expr = get_expression(match, skip_checks=skip_checks)
         value = context.select(expr, unwrap=True)
         validate_value(value, key)
-        buf += template[index:start] + to_str(value)
+        buf += template[index:start] + to_str(value, config=config)
         index = end
     buf += template[index:]
     # regex already backtracks and avoids any `${` starting with

dvc/render/match.py

@@ -1,3 +1,4 @@
+import logging
 import os
 from collections import defaultdict
 from typing import TYPE_CHECKING, DefaultDict, Dict, List, NamedTuple, Optional
@@ -17,6 +18,7 @@ if TYPE_CHECKING:
 
 
 dpath.options.ALLOW_EMPTY_STRING_KEYS = True
+logger = logging.getLogger(__name__)
 
 
 def _squash_plots_properties(data: List) -> Dict:
@@ -113,6 +115,7 @@ def match_defs_renderers(  # noqa: C901, PLR0912
             try:
                 dps, rev_props = converter.flat_datapoints(rev)
             except Exception as e:  # noqa: BLE001, pylint: disable=broad-except
+                logger.warning("In %r, %s", rev, str(e).lower())
                 def_errors[rev] = e
                 continue
 

dvc/repo/__init__.py

@@ -29,7 +29,7 @@ if TYPE_CHECKING:
     from dvc.stage import Stage
     from dvc.types import DictStrAny
     from dvc_data.hashfile.state import StateBase
-    from dvc_data.index import DataIndex
+    from dvc_data.index import DataIndex, DataIndexEntry
 
     from .experiments import Experiments
     from .index import Index
@@ -133,7 +133,7 @@ class Repo:
         assert root_dir
         return root_dir, dvc_dir
 
-    def __init__(  # noqa: PLR0915
+    def __init__(  # noqa: PLR0915, PLR0913
         self,
         root_dir: Optional[str] = None,
         fs: Optional["FileSystem"] = None,
@@ -144,9 +144,10 @@ class Repo:
         url: Optional[str] = None,
         repo_factory: Optional[Callable] = None,
         scm: Optional[Union["Git", "NoSCM"]] = None,
+        remote: Optional[str] = None,
+        remote_config: Optional["DictStrAny"] = None,
     ):
         from dvc.cachemgr import CacheManager
-        from dvc.config import Config
         from dvc.data_cloud import DataCloud
         from dvc.fs import GitFileSystem, LocalFileSystem, localfs
         from dvc.lock import LockNoop, make_lock
@@ -163,6 +164,9 @@ class Repo:
         self._fs_conf = {"repo_factory": repo_factory}
         self._fs = fs or localfs
         self._scm = scm
+        self._config = config
+        self._remote = remote
+        self._remote_config = remote_config
         self._data_index = None
 
         if rev and not fs:
@@ -182,7 +186,6 @@ class Repo:
             scm=scm,
         )
 
-        self.config: Config = Config(self.dvc_dir, fs=self.fs, config=config)
         self._uninitialized = uninitialized
 
         # used by DVCFileSystem to determine if it should traverse subrepos
@@ -234,6 +237,18 @@ class Repo:
     def __str__(self):
         return self.url or self.root_dir
 
+    @cached_property
+    def config(self):
+        from dvc.config import Config
+
+        return Config(
+            self.dvc_dir,
+            fs=self.fs,
+            config=self._config,
+            remote=self._remote,
+            remote_config=self._remote_config,
+        )
+
     @cached_property
     def local_dvc_dir(self):
         from dvc.fs import GitFileSystem, LocalFileSystem
@@ -285,7 +300,7 @@ class Repo:
             new.check_graph()
 
     @staticmethod
-    def open(url, *args, **kwargs):  # noqa: A003
+    def open(url: Optional[str], *args, **kwargs) -> "Repo":  # noqa: A003
         from .open_repo import open_repo
 
         return open_repo(url, *args, **kwargs)
@@ -375,6 +390,28 @@ class Repo:
         self.data_index.commit()
         self._reset()
 
+    def get_data_index_entry(
+        self,
+        path: str,
+        workspace: str = "repo",
+    ) -> Tuple["DataIndex", "DataIndexEntry"]:
+        if self.subrepos:
+            fs_path = self.dvcfs.from_os_path(path)
+            fs = self.dvcfs.fs
+            # pylint: disable-next=protected-access
+            key = fs._get_key_from_relative(fs_path)
+            # pylint: disable-next=protected-access
+            subrepo, _, key = fs._get_subrepo_info(key)
+            index = subrepo.index.data[workspace]
+        else:
+            index = self.index.data[workspace]
+            key = self.fs.path.relparts(path, self.root_dir)
+
+        try:
+            return index, index[key]
+        except KeyError as exc:
+            raise OutputNotFoundError(path, self) from exc
+
     def __repr__(self):
         return f"{self.__class__.__name__}: '{self.root_dir}'"
 
@@ -384,7 +421,7 @@ class Repo:
 
         fs = fs or localfs
         root = root or os.curdir
-        root_dir = fs.path.realpath(root)
+        root_dir = fs.path.abspath(root)
 
         if not fs.isdir(root_dir):
             raise NotDvcRepoError(f"directory '{root}' does not exist")
@@ -557,6 +594,7 @@ class Repo:
 
         from dvc.dirs import site_cache_dir
         from dvc.fs import GitFileSystem
+        from dvc.version import version_tuple
 
         cache_dir = self.config["core"].get("site_cache_dir") or site_cache_dir()
 
@@ -578,7 +616,13 @@ class Repo:
             os.umask(umask)
 
         md5 = hashlib.md5(  # noqa: S324  # nosec B324, B303
-            str((root_dir, getpass.getuser())).encode()
+            str(
+                (
+                    root_dir,
+                    getpass.getuser(),
+                    version_tuple[0],
+                )
+            ).encode()
         )
         repo_token = md5.hexdigest()
         return os.path.join(repos_dir, repo_token)
@@ -586,16 +630,21 @@ class Repo:
     def close(self):
         self.scm.close()
         self.state.close()
+        if "dvcfs" in self.__dict__:
+            self.dvcfs.close()
         if self._data_index is not None:
             self._data_index.close()
 
     def _reset(self):
         self.scm._reset()  # pylint: disable=protected-access
         self.state.close()
+        if "dvcfs" in self.__dict__:
+            self.dvcfs.close()
         self.__dict__.pop("index", None)
         self.__dict__.pop("dvcignore", None)
         self.__dict__.pop("dvcfs", None)
         self.__dict__.pop("datafs", None)
+        self.__dict__.pop("config", None)
 
     def __enter__(self):
         return self

dvc/repo/brancher.py

@@ -67,6 +67,7 @@ def brancher(  # noqa: E302
 
     saved_fs = self.fs
     saved_root = self.root_dir
+    saved_dvc_dir = self.dvc_dir
 
     scm = self.scm
 
@@ -101,6 +102,8 @@ def brancher(  # noqa: E302
     finally:
         self.fs = saved_fs
         self.root_dir = saved_root
+        self.dvc_dir = saved_dvc_dir
+        self._reset()  # pylint: disable=protected-access
 
 
 def _switch_fs(
@@ -125,6 +128,8 @@ def _switch_fs(
 
     repo.fs = fs
     repo.root_dir = root_dir
+    repo.dvc_dir = fs.path.join(root_dir, repo.DVC_DIR)
+    repo._reset()  # pylint: disable=protected-access
 
     if cwd_parts:
         cwd = repo.fs.path.join("/", *cwd_parts)
@@ -149,9 +154,12 @@ def switch(repo: "Repo", rev: str) -> Iterator[str]:
 
     saved_fs = repo.fs
     saved_root = repo.root_dir
+    saved_dvc_dir = repo.dvc_dir
     try:
         _switch_fs(repo, rev, repo_root_parts, cwd_parts)
         yield rev
     finally:
         repo.fs = saved_fs
         repo.root_dir = saved_root
+        repo.dvc_dir = saved_dvc_dir
+        repo._reset()  # pylint: disable=protected-access

dvc/repo/experiments/collect.py

@@ -1,6 +1,7 @@
 import itertools
 import logging
 import os
+from dataclasses import fields
 from datetime import datetime
 from typing import (
     TYPE_CHECKING,
@@ -152,7 +153,23 @@ def collect_queued(
     """
     if not baseline_revs:
         return {}
-    return repo.experiments.celery_queue.collect_queued_data(baseline_revs, **kwargs)
+    queued_data = {}
+    for rev, ranges in repo.experiments.celery_queue.collect_queued_data(
+        baseline_revs, **kwargs
+    ).items():
+        for exp_range in ranges:
+            for exp_state in exp_range.revs:
+                if exp_state.data:
+                    attrs = [f.name for f in fields(SerializableExp)]
+                    exp_state.data = SerializableExp(
+                        **{
+                            attr: getattr(exp_state.data, attr)
+                            for attr in attrs
+                            if attr != "metrics"
+                        }
+                    )
+        queued_data[rev] = ranges
+    return queued_data
 
 
 def collect_active(

dvc/repo/experiments/executor/base.py

@@ -44,7 +44,7 @@ if TYPE_CHECKING:
     from dvc.repo import Repo
     from dvc.repo.experiments.stash import ExpStashEntry
     from dvc.scm import Git
-    from dvc.stage import PipelineStage
+    from dvc.stage import PipelineStage, Stage
 
 logger = logging.getLogger(__name__)
 
@@ -255,6 +255,15 @@ class BaseExecutor(ABC):
             **kwargs,
         )
 
+    @classmethod
+    def _get_stage_files(cls, stages: List["Stage"]) -> List[str]:
+        from dvc.stage.utils import _get_stage_files
+
+        ret: List[str] = []
+        for stage in stages:
+            ret.extend(_get_stage_files(stage))
+        return ret
+
     @classmethod
     def _get_top_level_paths(cls, repo: "Repo") -> List["str"]:
         return list(
@@ -459,25 +468,16 @@ class BaseExecutor(ABC):
             should force overwrite any existing duplicates.
         """
         from dvc.repo.checkout import checkout as dvc_checkout
-        from dvc.repo.reproduce import reproduce as dvc_reproduce
-        from dvc.stage import PipelineStage
         from dvc.ui import ui
 
         auto_push = env2bool(DVC_EXP_AUTO_PUSH)
         git_remote = os.getenv(DVC_EXP_GIT_REMOTE, None)
 
-        unchanged = []
-
         if queue is not None:
             queue.put((rev, os.getpid()))
         if log_errors and log_level is not None:
             cls._set_log_level(log_level)
 
-        def filter_pipeline(stages):
-            unchanged.extend(
-                [stage for stage in stages if isinstance(stage, PipelineStage)]
-            )
-
         exp_hash: Optional[str] = None
         exp_ref: Optional["ExpRefInfo"] = None
         repro_force: bool = False
@@ -519,12 +519,10 @@ class BaseExecutor(ABC):
                     recursive=kwargs.get("recursive", False),
                 )
 
-            stages = dvc_reproduce(
-                dvc,
-                *args,
-                on_unchanged=filter_pipeline,
-                **kwargs,
-            )
+            stages = dvc.reproduce(*args, **kwargs)
+            if paths := cls._get_stage_files(stages):
+                logger.debug("Staging stage-related files: %s", paths)
+                dvc.scm_context.add(paths)
             if paths := cls._get_top_level_paths(dvc):
                 logger.debug("Staging top-level files: %s", paths)
                 dvc.scm_context.add(paths)
@@ -609,7 +607,7 @@ class BaseExecutor(ABC):
             old_cwd = os.getcwd()
 
             for path in copy_paths or []:
-                cls._copy_path(os.path.realpath(path), os.path.join(dvc.root_dir, path))
+                cls._copy_path(os.path.abspath(path), os.path.join(dvc.root_dir, path))
 
             if info.wdir:
                 os.chdir(os.path.join(dvc.scm.root_dir, info.wdir))

dvc/repo/experiments/queue/tempdir.py

@@ -115,7 +115,9 @@ class TempDirQueue(WorkspaceQueue):
                 message=message,
             )
             if not exec_result.exp_hash:
-                raise DvcException(f"Failed to reproduce experiment '{rev[:7]}'")
+                raise DvcException(  # noqa: TRY301
+                    f"Failed to reproduce experiment '{rev[:7]}'"
+                )
             if exec_result.ref_info:
                 results[rev].update(
                     self.collect_executor(self.repo.experiments, executor, exec_result)

dvc/repo/experiments/queue/workspace.py

@@ -121,7 +121,9 @@ class WorkspaceQueue(BaseStashQueue):
                 message=kwargs.get("message"),
             )
             if not exec_result.exp_hash:
-                raise DvcException(f"Failed to reproduce experiment '{rev[:7]}'")
+                raise DvcException(  # noqa: TRY301
+                    f"Failed to reproduce experiment '{rev[:7]}'"
+                )
             if exec_result.ref_info:
                 results[rev].update(
                     self.collect_executor(self.repo.experiments, executor, exec_result)

dvc/repo/experiments/utils.py

@@ -234,7 +234,7 @@ def fix_exp_head(scm: Union["Git", "NoSCM"], ref: Optional[str]) -> Optional[str
     if ref:
         name, tail = Git.split_ref_pattern(ref)
         if name == "HEAD" and scm.get_ref(EXEC_BASELINE):
-            return "".join((EXEC_BASELINE, tail))
+            return f"{EXEC_BASELINE}{tail}"
     return ref
 
 

dvc/repo/fetch.py

@@ -1,12 +1,64 @@
 import logging
+from typing import List, Tuple
 
 from dvc.exceptions import DownloadError
+from dvc_data.index import DataIndex, FileStorage
 
 from . import locked
 
 logger = logging.getLogger(__name__)
 
 
+def _collect_indexes(  # noqa: PLR0913
+    repo,
+    targets=None,
+    remote=None,
+    all_branches=False,
+    with_deps=False,
+    all_tags=False,
+    recursive=False,
+    all_commits=False,
+    revs=None,
+    max_size=None,
+    types=None,
+    config=None,
+):
+    indexes = {}
+    collection_exc = None
+
+    config = config or {}
+    if remote:
+        core = config.get("core") or {}
+        core["remote"] = remote
+        config["core"] = core
+
+    for rev in repo.brancher(
+        revs=revs,
+        all_branches=all_branches,
+        all_tags=all_tags,
+        all_commits=all_commits,
+    ):
+        try:
+            repo.config.update(config)
+
+            idx = repo.index.targets_view(
+                targets,
+                with_deps=with_deps,
+                recursive=recursive,
+                max_size=max_size,
+                types=types,
+            )
+            indexes[idx.data_tree.hash_info.value] = idx.data["repo"]
+        except Exception as exc:  # pylint: disable=broad-except
+            collection_exc = exc
+            logger.exception("failed to collect '%s'", rev or "workspace")
+
+    if not indexes and collection_exc:
+        raise collection_exc
+
+    return indexes
+
+
 @locked
 def fetch(  # noqa: C901, PLR0913
     self,
@@ -20,6 +72,9 @@ def fetch(  # noqa: C901, PLR0913
     all_commits=False,
     run_cache=False,
     revs=None,
+    max_size=None,
+    types=None,
+    config=None,
 ) -> int:
     """Download data items from a cloud and imported repositories
 
@@ -51,39 +106,36 @@ def fetch(  # noqa: C901, PLR0913
     except DownloadError as exc:
         failed_count += exc.amount
 
-    indexes = []
-    index_keys = set()
-    for _ in self.brancher(
-        revs=revs,
+    indexes = _collect_indexes(
+        self,
+        targets=targets,
+        remote=remote,
         all_branches=all_branches,
+        with_deps=with_deps,
         all_tags=all_tags,
+        recursive=recursive,
         all_commits=all_commits,
-    ):
-        saved_remote = self.config["core"].get("remote")
-        try:
-            if remote:
-                self.config["core"]["remote"] = remote
+        revs=revs,
+        max_size=max_size,
+        types=types,
+        config=config,
+    )
 
-            idx = self.index.targets_view(
-                targets,
-                with_deps=with_deps,
-                recursive=recursive,
-            )
-            index_keys.add(idx.data_tree.hash_info.value)
-            indexes.append(idx.data["repo"])
-        finally:
-            if remote:
-                self.config["core"]["remote"] = saved_remote
-
-    cache_key = ("fetch", tokenize(sorted(index_keys)))
+    cache_key = ("fetch", tokenize(sorted(indexes.keys())))
 
     with Callback.as_tqdm_callback(
         desc="Collecting",
         unit="entry",
     ) as cb:
         data = collect(
-            indexes, cache_index=self.data_index, cache_key=cache_key, callback=cb
+            indexes.values(),
+            "remote",
+            cache_index=self.data_index,
+            cache_key=cache_key,
+            callback=cb,
         )
+    data, unversioned_count = _log_unversioned(data)
+    failed_count += unversioned_count
 
     with Callback.as_tqdm_callback(
         desc="Fetching",
@@ -96,7 +148,7 @@ def fetch(  # noqa: C901, PLR0913
                 callback=cb,
             )  # pylint: disable=assignment-from-no-return
         finally:
-            for fs_index in data.values():
+            for fs_index in data:
                 fs_index.close()
 
     if fetch_transferred:
@@ -109,3 +161,34 @@ def fetch(  # noqa: C901, PLR0913
         raise DownloadError(failed_count)
 
     return transferred_count
+
+
+def _log_unversioned(data: List["DataIndex"]) -> Tuple[List["DataIndex"], int]:
+    ret: List["DataIndex"] = []
+    unversioned: List[str] = []
+    for fs_index in data:
+        remote = fs_index.storage_map[()].remote
+        if not isinstance(remote, FileStorage) or not remote.fs.version_aware:
+            ret.append(fs_index)
+            continue
+
+        fs = remote.fs
+        index = DataIndex()
+        index.storage_map = fs_index.storage_map
+        for key, entry in fs_index.iteritems():
+            if entry.meta and not entry.meta.isdir and entry.meta.version_id is None:
+                unversioned.append(fs.unstrip_protocol(fs.path.join(remote.path, *key)))
+            else:
+                index[key] = entry
+        fs_index.close()
+        ret.append(index)
+
+    if unversioned:
+        logger.warning(
+            (
+                "Some files are missing cloud version information and will not be "
+                "fetched from the remote:\n%s"
+            ),
+            "\n".join(unversioned),
+        )
+    return ret, len(unversioned)

dvc/repo/get.py

@@ -19,22 +19,39 @@ class GetDVCFileError(DvcException):
         )
 
 
-def get(url, path, out=None, rev=None, jobs=None, force=False):
+def get(
+    url,
+    path,
+    out=None,
+    rev=None,
+    jobs=None,
+    force=False,
+    config=None,
+    remote=None,
+    remote_config=None,
+):
+    from dvc.config import Config
     from dvc.dvcfile import is_valid_filename
-    from dvc.fs.callbacks import Callback
     from dvc.repo import Repo
 
     out = resolve_output(path, out, force=force)
 
     if is_valid_filename(out):
-        raise GetDVCFileError()
+        raise GetDVCFileError
+
+    if config and not isinstance(config, dict):
+        config = Config.load_file(config)
 
     with Repo.open(
         url=url,
         rev=rev,
         subrepos=True,
         uninitialized=True,
+        config=config,
+        remote=remote,
+        remote_config=remote_config,
     ) as repo:
+        from dvc.fs import download
         from dvc.fs.data import DataFileSystem
 
         fs: Union[DataFileSystem, "DVCFileSystem"]
@@ -44,14 +61,9 @@ def get(url, path, out=None, rev=None, jobs=None, force=False):
         else:
             fs = repo.dvcfs
             fs_path = fs.from_os_path(path)
-
-        with Callback.as_tqdm_callback(
-            desc=f"Downloading {fs.path.name(path)}",
-            unit="files",
-        ) as cb:
-            fs.get(
-                fs_path,
-                os.path.abspath(out),
-                batch_size=jobs,
-                callback=cb,
-            )
+        download(
+            fs,
+            fs_path,
+            os.path.abspath(out),
+            jobs=jobs,
+        )

dvc/repo/get_url.py

@@ -14,4 +14,4 @@ def get_url(url, out=None, *, config=None, jobs=None, force=False):
     src_fs, src_path = parse_external_url(url, config)
     if not src_fs.exists(src_path):
         raise URLMissingError(url)
-    download(src_fs, src_path, out, jobs=jobs)
+    download(src_fs, src_path, out.fs_path, jobs=jobs)

dvc/repo/graph.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, Iterator, List, Set
+from typing import TYPE_CHECKING, Any, Iterator, List, Optional, Set, TypeVar
 
 from dvc.fs import localfs
 from dvc.utils.fs import path_isin
@@ -8,6 +8,8 @@ if TYPE_CHECKING:
 
     from dvc.stage import Stage
 
+T = TypeVar("T")
+
 
 def check_acyclic(graph: "DiGraph") -> None:
     import networkx as nx
@@ -29,6 +31,9 @@ def check_acyclic(graph: "DiGraph") -> None:
 
 def get_pipeline(pipelines, node):
     found = [i for i in pipelines if i.has_node(node)]
+    if not found:
+        return None
+
     assert len(found) == 1
     return found[0]
 
@@ -39,10 +44,28 @@ def get_pipelines(graph: "DiGraph"):
     return [graph.subgraph(c).copy() for c in nx.weakly_connected_components(graph)]
 
 
+def get_subgraph_of_nodes(
+    graph: "DiGraph", sources: Optional[List[Any]] = None, downstream: bool = False
+) -> "DiGraph":
+    from networkx import dfs_postorder_nodes, reverse_view
+
+    if not sources:
+        return graph
+
+    g = reverse_view(graph) if downstream else graph
+    nodes = []
+    for source in sources:
+        nodes.extend(dfs_postorder_nodes(g, source))
+    return graph.subgraph(nodes)
+
+
 def collect_pipeline(stage: "Stage", graph: "DiGraph") -> Iterator["Stage"]:
     import networkx as nx
 
     pipeline = get_pipeline(get_pipelines(graph), stage)
+    if not pipeline:
+        return iter([])
+
     return nx.dfs_postorder_nodes(pipeline, stage)
 
 

dvc/repo/imp.py

@@ -1,6 +1,32 @@
-def imp(self, url, path, out=None, rev=None, **kwargs):
+def imp(
+    self,
+    url,
+    path,
+    out=None,
+    rev=None,
+    config=None,
+    remote=None,
+    remote_config=None,
+    **kwargs,
+):
     erepo = {"url": url}
     if rev is not None:
         erepo["rev"] = rev
 
+    if config is not None:
+        erepo["config"] = config
+
+    if remote is not None and remote_config is not None:
+        conf = erepo.get("config") or {}
+        remotes = conf.get("remote") or {}
+        remote_conf = remotes.get(remote) or {}
+        remote_conf.update(remote_config)
+        remotes[remote] = remote_conf
+        conf["remote"] = remotes
+        erepo["config"] = conf
+    elif remote is not None:
+        erepo["remote"] = remote
+    elif remote_config is not None:
+        erepo["remote"] = remote_config
+
     return self.imp_url(path, out=out, erepo=erepo, frozen=True, **kwargs)

dvc/repo/index.py

@@ -374,6 +374,50 @@ class Index:
 
         return dict(by_workspace)
 
+    @cached_property
+    def metric_keys(self) -> Dict[str, Set["DataIndexKey"]]:
+        from collections import defaultdict
+
+        from .metrics.show import _collect_top_level_metrics
+
+        by_workspace: Dict[str, Set["DataIndexKey"]] = defaultdict(set)
+
+        by_workspace["repo"] = set()
+
+        for out in self.outs:
+            if not out.metric:
+                continue
+
+            workspace, key = out.index_key
+            by_workspace[workspace].add(key)
+
+        for path in _collect_top_level_metrics(self.repo):
+            key = self.repo.fs.path.relparts(path, self.repo.root_dir)
+            by_workspace["repo"].add(key)
+
+        return dict(by_workspace)
+
+    @cached_property
+    def plot_keys(self) -> Dict[str, Set["DataIndexKey"]]:
+        from collections import defaultdict
+
+        by_workspace: Dict[str, Set["DataIndexKey"]] = defaultdict(set)
+
+        by_workspace["repo"] = set()
+
+        for out in self.outs:
+            if not out.plot:
+                continue
+
+            workspace, key = out.index_key
+            by_workspace[workspace].add(key)
+
+        for path in self._plot_sources:
+            key = self.repo.fs.path.relparts(path, self.repo.root_dir)
+            by_workspace["repo"].add(key)
+
+        return dict(by_workspace)
+
     @cached_property
     def data_tree(self):
         from dvc_data.hashfile.tree import Tree
@@ -487,11 +531,31 @@ class Index:
                 used[odb].update(objs)
         return used
 
+    def _types_filter(self, types, out):
+        ws, okey = out.index_key
+        for typ in types:
+            if typ == "plots":
+                keys = self.plot_keys
+            elif typ == "metrics":
+                keys = self.metric_keys
+            else:
+                raise ValueError(f"unsupported type {typ}")
+
+            for key in keys.get(ws, []):
+                if (len(key) >= len(okey) and key[: len(okey)] == okey) or (
+                    len(key) < len(okey) and okey[: len(key)] == key
+                ):
+                    return True
+
+        return False
+
     def targets_view(
         self,
         targets: Optional["TargetType"],
         stage_filter: Optional[Callable[["Stage"], bool]] = None,
         outs_filter: Optional[Callable[["Output"], bool]] = None,
+        max_size: Optional[int] = None,
+        types: Optional[List[str]] = None,
         **kwargs: Any,
     ) -> "IndexView":
         """Return read-only view of index for the specified targets.
@@ -514,7 +578,20 @@ class Index:
             for stage_info in self.collect_targets(targets, **kwargs)
             if not stage_filter or stage_filter(stage_info.stage)
         ]
-        return IndexView(self, stage_infos, outs_filter=outs_filter)
+
+        def _outs_filter(out):
+            if max_size and out.meta and out.meta.size and out.meta.size >= max_size:
+                return False
+
+            if types and not self._types_filter(types, out):
+                return False
+
+            if outs_filter:
+                return outs_filter(out)
+
+            return True
+
+        return IndexView(self, stage_infos, outs_filter=_outs_filter)
 
 
 class _DataPrefixes(NamedTuple):

dvc/repo/init.py

@@ -37,7 +37,7 @@ def init(root_dir=os.curdir, no_scm=False, force=False, subdir=False):  # noqa:
             "Cannot initialize repo with `--no-scm` and `--subdir`"
         )
 
-    root_dir = os.path.realpath(root_dir)
+    root_dir = os.path.abspath(root_dir)
     dvc_dir = os.path.join(root_dir, Repo.DVC_DIR)
 
     try:

dvc/repo/ls.py

@@ -13,6 +13,9 @@ def ls(
     rev: Optional[str] = None,
     recursive: Optional[bool] = None,
     dvc_only: bool = False,
+    config: Optional[str] = None,
+    remote: Optional[str] = None,
+    remote_config: Optional[dict] = None,
 ):
     """Methods for getting files and outputs for the repo.
 
@@ -22,6 +25,9 @@ def ls(
         rev (str, optional): SHA commit, branch or tag name
         recursive (bool, optional): recursively walk the repo
         dvc_only (bool, optional): show only DVC-artifacts
+        config (str, optional): path to config file
+        remote (str, optional): remote name to set as a default remote in the repo
+        remote_config (str, dict): remote config to merge with a remote in the repo
 
     Returns:
         list of `entry`
@@ -35,9 +41,24 @@ def ls(
             "isexec": bool,
         }
     """
+    from dvc.config import Config
+
     from . import Repo
 
-    with Repo.open(url, rev=rev, subrepos=True, uninitialized=True) as repo:
+    if config and not isinstance(config, dict):
+        config_dict = Config.load_file(config)
+    else:
+        config_dict = None
+
+    with Repo.open(
+        url,
+        rev=rev,
+        subrepos=True,
+        uninitialized=True,
+        config=config_dict,
+        remote=remote,
+        remote_config=remote_config,
+    ) as repo:
         path = path or ""
 
         ret = _ls(repo, path, recursive, dvc_only)

dvc/repo/open_repo.py

@@ -2,8 +2,7 @@ import logging
 import os
 import tempfile
 import threading
-from contextlib import contextmanager
-from typing import TYPE_CHECKING, Dict, Iterator, Optional, Tuple
+from typing import TYPE_CHECKING, Dict, Optional, Tuple
 
 from funcy import retry, wrap_with
 
@@ -18,16 +17,14 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
-@contextmanager
 @map_scm_exception()
 def _external_repo(
     url,
     rev: Optional[str] = None,
-    for_write: bool = False,
     **kwargs,
-) -> Iterator["Repo"]:
+) -> "Repo":
     logger.debug("Creating external repo %s@%s", url, rev)
-    path = _cached_clone(url, rev, for_write=for_write)
+    path = _cached_clone(url, rev)
     # Local HEAD points to the tip of whatever branch we first cloned from
     # (which may not be the default branch), use origin/HEAD here to get
     # the tip of the default branch
@@ -38,11 +35,6 @@ def _external_repo(
     config.update(kwargs.pop("config", None) or {})
 
     main_root = "/"
-    if for_write:
-        # we already checked out needed revision
-        rev = None
-        main_root = path
-
     repo_kwargs = dict(
         root_dir=path,
         url=url,
@@ -52,14 +44,7 @@ def _external_repo(
         **kwargs,
     )
 
-    repo = Repo(**repo_kwargs)
-
-    try:
-        yield repo
-    finally:
-        repo.close()
-        if for_write:
-            _remove(path)
+    return Repo(**repo_kwargs)
 
 
 def open_repo(url, *args, **kwargs):
@@ -67,6 +52,7 @@ def open_repo(url, *args, **kwargs):
         url = os.getcwd()
 
     if os.path.exists(url):
+        url = os.path.abspath(url)
         try:
             config = _get_remote_config(url)
             config.update(kwargs.get("config") or {})
@@ -139,20 +125,19 @@ def _get_remote_config(url):
         repo.close()
 
 
-def _cached_clone(url, rev, for_write=False):
+def _cached_clone(url, rev):
     """Clone an external git repo to a temporary directory.
 
     Returns the path to a local temporary directory with the specified
-    revision checked out. If for_write is set prevents reusing this dir via
-    cache.
+    revision checked out.
     """
     from shutil import copytree
 
     # even if we have already cloned this repo, we may need to
     # fetch/fast-forward to get specified rev
-    clone_path, shallow = _clone_default_branch(url, rev, for_write=for_write)
+    clone_path, shallow = _clone_default_branch(url, rev)
 
-    if not for_write and (url) in CLONES:
+    if url in CLONES:
         return CLONES[url][0]
 
     # Copy to a new dir to keep the clone clean
@@ -160,16 +145,12 @@ def _cached_clone(url, rev, for_write=False):
     logger.debug("erepo: making a copy of %s clone", url)
     copytree(clone_path, repo_path)
 
-    # Check out the specified revision
-    if for_write:
-        _git_checkout(repo_path, rev)
-    else:
-        CLONES[url] = (repo_path, shallow)
+    CLONES[url] = (repo_path, shallow)
     return repo_path
 
 
 @wrap_with(threading.Lock())
-def _clone_default_branch(url, rev, for_write=False):  # noqa: C901, PLR0912
+def _clone_default_branch(url, rev):  # noqa: C901, PLR0912
     """Get or create a clean clone of the url.
 
     The cloned is reactualized with git pull unless rev is a known sha.
@@ -203,7 +184,7 @@ def _clone_default_branch(url, rev, for_write=False):  # noqa: C901, PLR0912
 
             logger.debug("erepo: git clone '%s' to a temporary dir", url)
             clone_path = tempfile.mkdtemp("dvc-clone")
-            if not for_write and rev and not Git.is_sha(rev):
+            if rev and not Git.is_sha(rev):
                 # If rev is a tag or branch name try shallow clone first
 
                 try:
@@ -248,17 +229,6 @@ def _merge_upstream(git: "Git"):
         pass
 
 
-def _git_checkout(repo_path, rev):
-    from dvc.scm import Git
-
-    logger.debug("erepo: git checkout %s@%s", repo_path, rev)
-    git = Git(repo_path)
-    try:
-        git.checkout(rev)
-    finally:
-        git.close()
-
-
 def _remove(path):
     from dvc.utils.fs import remove
 

dvc/repo/plots/__init__.py

@@ -134,7 +134,7 @@ class Plots:
             if "workspace" in revs:
                 # reorder revs to match repo.brancher ordering
                 revs.remove("workspace")
-                revs = ["workspace"] + revs
+                revs = ["workspace", *revs]
         for rev in revs:
             with switch_repo(self.repo, rev) as (repo, _):
                 res: Dict = {}
@@ -483,7 +483,7 @@ def _collect_pipeline_files(repo, targets: List[str], props, onerror=None):
             if isinstance(elem, str):
                 dvcfile_defs_dict[elem] = None
             else:
-                k, v = list(elem.items())[0]
+                k, v = next(iter(elem.items()))
                 dvcfile_defs_dict[k] = v
 
         resolved = _resolve_definitions(

dvc/repo/pull.py

@@ -1,5 +1,6 @@
 import logging
 
+from dvc.exceptions import CheckoutError
 from dvc.repo import locked
 from dvc.utils import glob_targets
 
@@ -38,13 +39,17 @@ def pull(  # noqa: PLR0913
         recursive=recursive,
         run_cache=run_cache,
     )
-    stats = self.checkout(
-        targets=expanded_targets,
-        with_deps=with_deps,
-        force=force,
-        recursive=recursive,
-        allow_missing=allow_missing,
-    )
+    try:
+        stats = self.checkout(
+            targets=expanded_targets,
+            with_deps=with_deps,
+            force=force,
+            recursive=recursive,
+            allow_missing=allow_missing,
+        )
+    except CheckoutError as exc:
+        exc.stats["fetched"] = processed_files_count
+        raise
 
     stats["fetched"] = processed_files_count
     return stats

dvc/repo/reproduce.py

@@ -1,130 +1,85 @@
 import logging
-from typing import TYPE_CHECKING, Iterator, List
+from typing import (
+    TYPE_CHECKING,
+    Callable,
+    Dict,
+    Iterable,
+    List,
+    NoReturn,
+    Optional,
+    Set,
+    Tuple,
+    TypeVar,
+    Union,
+    cast,
+)
+
+from funcy import ldistinct
 
 from dvc.exceptions import ReproductionError
 from dvc.repo.scm_context import scm_context
 from dvc.stage.cache import RunCacheNotSupported
+from dvc.utils import humanize
+from dvc.utils.collections import ensure_list
 
 from . import locked
 
 if TYPE_CHECKING:
+    from networkx import DiGraph
+
     from dvc.stage import Stage
 
     from . import Repo
 
 logger = logging.getLogger(__name__)
+T = TypeVar("T")
 
 
-def _reproduce_stage(stage: "Stage", **kwargs) -> List["Stage"]:
-    if stage.frozen and not stage.is_import:
-        logger.warning(
-            "%s is frozen. Its dependencies are not going to be reproduced.",
-            stage,
-        )
-
-    stage = stage.reproduce(**kwargs)
-    if not stage:
-        return []
-
-    if not kwargs.get("dry", False):
-        stage.dump(update_pipeline=False)
-        _track_stage(stage)
-
-    return [stage]
+def collect_stages(
+    repo: "Repo",
+    targets: Iterable[str],
+    recursive: bool = False,
+    glob: bool = False,
+) -> List["Stage"]:
+    stages: List["Stage"] = []
+    for target in targets:
+        stages.extend(repo.stage.collect(target, recursive=recursive, glob=glob))
+    return ldistinct(stages)
 
 
-def _get_stage_files(stage: "Stage") -> Iterator[str]:
-    yield stage.dvcfile.relpath
-    for dep in stage.deps:
-        if (
-            not dep.use_scm_ignore
-            and dep.is_in_repo
-            and not stage.repo.dvcfs.isdvc(stage.repo.dvcfs.from_os_path(str(dep)))
-        ):
-            yield dep.fs_path
-    for out in stage.outs:
-        if not out.use_scm_ignore and out.is_in_repo:
-            yield out.fs_path
+def get_subgraph(
+    graph: "DiGraph",
+    nodes: Optional[List] = None,
+    pipeline: bool = False,
+    downstream: bool = False,
+) -> "DiGraph":
+    import networkx as nx
+
+    from .graph import get_pipeline, get_pipelines, get_subgraph_of_nodes
+
+    if not pipeline or not nodes:
+        return get_subgraph_of_nodes(graph, nodes, downstream=downstream)
+
+    pipelines = get_pipelines(graph)
+    used_pipelines = [get_pipeline(pipelines, node) for node in nodes]
+    return nx.compose_all(used_pipelines)
 
 
-def _track_stage(stage: "Stage") -> None:
-    from dvc.utils import relpath
-
-    context = stage.repo.scm_context
-    for path in _get_stage_files(stage):
-        context.track_file(relpath(path))
-    return context.track_changed_files()
+def get_active_graph(graph: "DiGraph") -> "DiGraph":
+    g = cast("DiGraph", graph.copy())
+    for stage in graph:
+        if stage.frozen:
+            # NOTE: disconnect frozen stage from its dependencies
+            g.remove_edges_from(graph.out_edges(stage))
+    return g
 
 
-@locked
-@scm_context
-def reproduce(  # noqa: C901, PLR0912
-    self: "Repo",
-    targets=None,
-    recursive=False,
-    pipeline=False,
-    all_pipelines=False,
-    **kwargs,
-):
-    from .graph import get_pipeline, get_pipelines
-
-    glob = kwargs.pop("glob", False)
-
-    if isinstance(targets, str):
-        targets = [targets]
-
-    if not all_pipelines and not targets:
-        from dvc.dvcfile import PROJECT_FILE
-
-        targets = [PROJECT_FILE]
-
-    interactive = kwargs.get("interactive", False)
-    if not interactive:
-        kwargs["interactive"] = self.config["core"].get("interactive", False)
-
-    stages = set()
-    if pipeline or all_pipelines:
-        pipelines = get_pipelines(self.index.graph)
-        if all_pipelines:
-            used_pipelines = pipelines
-        else:
-            used_pipelines = []
-            for target in targets:
-                stage = self.stage.get_target(target)
-                used_pipelines.append(get_pipeline(pipelines, stage))
-
-        for pline in used_pipelines:
-            for stage in pline:
-                if pline.in_degree(stage) == 0:
-                    stages.add(stage)
-    else:
-        for target in targets:
-            stages.update(
-                self.stage.collect(
-                    target,
-                    recursive=recursive,
-                    glob=glob,
-                )
-            )
-
-    if kwargs.get("pull", False) and kwargs.get("run_cache", True):
-        logger.debug("Pulling run cache")
-        try:
-            self.stage_cache.pull(None)
-        except RunCacheNotSupported as e:
-            logger.warning("Failed to pull run cache: %s", e)
-
-    return _reproduce_stages(self.index.graph, list(stages), **kwargs)
-
-
-def _reproduce_stages(  # noqa: C901
-    graph,
-    stages,
-    downstream=False,
-    single_item=False,
-    on_unchanged=None,
-    **kwargs,
-):
+def plan_repro(
+    graph: "DiGraph",
+    stages: Optional[List["T"]] = None,
+    pipeline: bool = False,
+    downstream: bool = False,
+) -> List["T"]:
     r"""Derive the evaluation of the given node for the given graph.
 
     When you _reproduce a stage_, you want to _evaluate the descendants_
@@ -160,80 +115,146 @@ def _reproduce_stages(  # noqa: C901
 
     The derived evaluation of _downstream_ B would be: [B, D, E]
     """
-    steps = _get_steps(graph, stages, downstream, single_item)
+    import networkx as nx
+
+    sub = get_subgraph(graph, stages, pipeline=pipeline, downstream=downstream)
+    return list(nx.dfs_postorder_nodes(sub))
+
+
+def _reproduce_stage(stage: "Stage", **kwargs) -> Optional["Stage"]:
+    if stage.frozen and not stage.is_import:
+        msg = "%s is frozen. Its dependencies are not going to be reproduced."
+        logger.warning(msg, stage)
+
+    ret = stage.reproduce(**kwargs)
+    if ret and not kwargs.get("dry", False):
+        stage.dump(update_pipeline=False)
+    return ret
+
+
+def _get_upstream_downstream_nodes(
+    graph: Optional["DiGraph"], node: T
+) -> Tuple[List[T], List[T]]:
+    succ = list(graph.successors(node)) if graph else []
+    pre = list(graph.predecessors(node)) if graph else []
+    return succ, pre
+
+
+def _repr(stages: Iterable["Stage"]) -> str:
+    return humanize.join(repr(stage.addressing) for stage in stages)
+
+
+def handle_error(
+    graph: Optional["DiGraph"], on_error: str, exc: Exception, stage: "Stage"
+) -> Set["Stage"]:
+    import networkx as nx
+
+    logger.warning("%s%s", exc, " (ignored)" if on_error == "ignore" else "")
+    if not graph or on_error == "ignore":
+        return set()
+
+    dependents = set(nx.dfs_postorder_nodes(graph.reverse(), stage)) - {stage}
+    if dependents:
+        names = _repr(dependents)
+        msg = "%s %s will be skipped due to this failure"
+        logger.warning(msg, "Stages" if len(dependents) > 1 else "Stage", names)
+    return dependents
+
+
+def _raise_error(exc: Optional[Exception], *stages: "Stage") -> NoReturn:
+    names = _repr(stages)
+    segment = " stages:" if len(stages) > 1 else ""
+    raise ReproductionError(f"failed to reproduce{segment} {names}") from exc
+
+
+def _reproduce(
+    stages: List["Stage"],
+    graph: Optional["DiGraph"] = None,
+    force_downstream: bool = False,
+    on_error: str = "fail",
+    force: bool = False,
+    repro_fn: Callable = _reproduce_stage,
+    **kwargs,
+) -> List["Stage"]:
+    assert on_error in ("fail", "keep-going", "ignore")
 
-    force_downstream = kwargs.pop("force_downstream", False)
     result: List["Stage"] = []
-    unchanged: List["Stage"] = []
-    # `ret` is used to add a cosmetic newline.
-    ret: List["Stage"] = []
+    failed: List["Stage"] = []
+    to_skip: Dict["Stage", "Stage"] = {}
+    ret: Optional["Stage"] = None
+
+    force_state = {node: force for node in stages}
+
+    for stage in stages:
+        if stage in to_skip:
+            continue
 
-    for stage in steps:
         if ret:
-            logger.info("")
+            logger.info("")  # add a newline
+
+        upstream, downstream = _get_upstream_downstream_nodes(graph, stage)
+        force_stage = force_state[stage]
 
         try:
-            kwargs["upstream"] = result + unchanged
-            ret = _reproduce_stage(stage, **kwargs)
+            ret = repro_fn(stage, upstream=upstream, force=force_stage, **kwargs)
+        except Exception as exc:  # noqa: BLE001, pylint: disable=broad-exception-caught
+            failed.append(stage)
+            if on_error == "fail":
+                _raise_error(exc, stage)
 
-            if len(ret) == 0:
-                unchanged.extend([stage])
-            elif force_downstream:
-                # NOTE: we are walking our pipeline from the top to the
-                # bottom. If one stage is changed, it will be reproduced,
-                # which tells us that we should force reproducing all of
-                # the other stages down below, even if their direct
-                # dependencies didn't change.
-                kwargs["force"] = True
+            dependents = handle_error(graph, on_error, exc, stage)
+            to_skip.update({node: stage for node in dependents})
+            continue
 
-            if ret:
-                result.extend(ret)
-        except Exception as exc:  # noqa: BLE001
-            raise ReproductionError(stage.addressing) from exc
+        if force_downstream and (ret or force_stage):
+            force_state.update({node: True for node in downstream})
 
-    if on_unchanged is not None:
-        on_unchanged(unchanged)
+        if ret:
+            result.append(ret)
+
+    if on_error != "ignore" and failed:
+        _raise_error(None, *failed)
     return result
 
 
-def _get_steps(graph, stages, downstream, single_item):
-    import networkx as nx
+@locked
+@scm_context
+def reproduce(
+    self: "Repo",
+    targets: Union[Iterable[str], str, None] = None,
+    recursive: bool = False,
+    pipeline: bool = False,
+    all_pipelines: bool = False,
+    downstream: bool = False,
+    single_item: bool = False,
+    glob: bool = False,
+    on_error: Optional[str] = "fail",
+    **kwargs,
+):
+    from dvc.dvcfile import PROJECT_FILE
 
-    active = graph.copy()
+    if all_pipelines or pipeline:
+        single_item = False
+        downstream = False
+
+    if not kwargs.get("interactive", False):
+        kwargs["interactive"] = self.config["core"].get("interactive", False)
+
+    stages: List["Stage"] = []
+    if not all_pipelines:
+        targets_list = ensure_list(targets or PROJECT_FILE)
+        stages = collect_stages(self, targets_list, recursive=recursive, glob=glob)
+
+    if kwargs.get("pull", False) and kwargs.get("run_cache", True):
+        logger.debug("Pulling run cache")
+        try:
+            self.stage_cache.pull(None)
+        except RunCacheNotSupported as e:
+            logger.warning("Failed to pull run cache: %s", e)
+
+    graph = None
+    steps = stages
     if not single_item:
-        # NOTE: frozen stages don't matter for single_item
-        for stage in graph:
-            if stage.frozen:
-                # NOTE: disconnect frozen stage from its dependencies
-                active.remove_edges_from(graph.out_edges(stage))
-
-    all_pipelines: List["Stage"] = []
-    for stage in stages:
-        if downstream:
-            # NOTE (py3 only):
-            # Python's `deepcopy` defaults to pickle/unpickle the object.
-            # Stages are complex objects (with references to `repo`,
-            # `outs`, and `deps`) that cause struggles when you try
-            # to serialize them. We need to create a copy of the graph
-            # itself, and then reverse it, instead of using
-            # graph.reverse() directly because it calls `deepcopy`
-            # underneath -- unless copy=False is specified.
-            nodes = nx.dfs_postorder_nodes(active.reverse(copy=False), stage)
-            all_pipelines += reversed(list(nodes))
-        else:
-            all_pipelines += nx.dfs_postorder_nodes(active, stage)
-
-    steps = []
-    for stage in all_pipelines:
-        if stage not in steps:
-            # NOTE: order of steps still matters for single_item
-            if single_item and stage not in stages:
-                continue
-
-            steps.append(stage)
-
-    return steps
-
-
-def _repro_callback(experiments_callback, unchanged, stages):
-    experiments_callback(unchanged, stages)
+        graph = get_active_graph(self.index.graph)
+        steps = plan_repro(graph, stages, pipeline=pipeline, downstream=downstream)
+    return _reproduce(steps, graph=graph, on_error=on_error or "fail", **kwargs)

dvc/repo/stage.py

@@ -37,11 +37,19 @@ StageSet = Set["Stage"]
 
 
 def _collect_with_deps(stages: StageList, graph: "DiGraph") -> StageSet:
+    from dvc.exceptions import StageNotFoundError
     from dvc.repo.graph import collect_pipeline
 
     res: StageSet = set()
     for stage in stages:
-        res.update(collect_pipeline(stage, graph=graph))
+        pl = list(collect_pipeline(stage, graph=graph))
+        if not pl:
+            raise StageNotFoundError(
+                f"Stage {stage} is not found in the project. "
+                "Check that there are no symlinks in the parents "
+                "leading up to it within the project."
+            )
+        res.update(pl)
     return res
 
 
@@ -211,7 +219,7 @@ class StageLoad:
         self, path: Optional[str] = None, name: Optional[str] = None
     ) -> str:
         if path:
-            return self.repo.fs.path.realpath(path)
+            return self.repo.fs.path.abspath(path)
 
         path = PROJECT_FILE
         logger.debug("Assuming '%s' to be a stage inside '%s'", name, path)

dvc/repo/worktree.py

@@ -427,7 +427,7 @@ def _get_update_diff_index(
         meta_cmp_key=partial(_meta_checksum, remote.fs),
         with_unchanged=True,
     ):
-        if change.typ == ADD or change.typ == MODIFY:
+        if change.typ in (ADD, MODIFY):
             entry = change.new
             # preserve md5's which were calculated in out.save() after
             # downloading

dvc/rwlock.py

@@ -31,15 +31,13 @@ RWLOCK_LOCK = "rwlock.lock"
 class RWLockFileCorruptedError(DvcException):
     def __init__(self, path):
         super().__init__(
-            "Unable to read RWLock-file '{}'. JSON structure is corrupted".format(
-                relpath(path)
-            )
+            f"Unable to read RWLock-file {relpath(path)!r}. JSON structure is corrupted"
         )
 
 
 class RWLockFileFormatError(DvcException):
     def __init__(self, path):
-        super().__init__(f"RWLock-file '{relpath(path)}' format error.")
+        super().__init__(f"RWLock-file {relpath(path)!r} format error.")
 
 
 @contextmanager

dvc/scm.py

@@ -67,7 +67,7 @@ def map_scm_exception(with_cause: bool = False) -> Iterator[None]:
         into = SCMError(str(exc))
         if with_cause:
             raise into from exc
-        raise into
+        raise into  # noqa: B904
 
 
 @overload

dvc/stage/__init__.py

@@ -426,10 +426,7 @@ class Stage(params.StageParams):
                 logger.info("Stage '%s' didn't change, skipping", self.addressing)
             return None
 
-        msg = "Going to reproduce {stage}. Are you sure you want to continue?".format(
-            stage=self
-        )
-
+        msg = f"Going to reproduce {self}. Are you sure you want to continue?"
         if interactive and not prompt.confirm(msg):
             raise DvcException("reproduction aborted by the user")
 

dvc/stage/cache.py

@@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, List, Optional, Tuple
 from funcy import first
 
 from dvc import fs
-from dvc.exceptions import DvcException
+from dvc.exceptions import CollectCacheError, DvcException
 from dvc.utils import dict_sha256, relpath
 
 if TYPE_CHECKING:
@@ -123,12 +123,15 @@ class StageCache:
 
     @contextmanager
     def _cache_type_copy(self):
-        cache_types = self.repo.cache.legacy.cache_types
+        cache_types = self.repo.cache.local.cache_types
+        legacy_cache_types = self.repo.cache.legacy.cache_types
+        self.repo.cache.local.cache_types = ["copy"]
         self.repo.cache.legacy.cache_types = ["copy"]
         try:
             yield
         finally:
-            self.repo.cache.legacy.cache_types = cache_types
+            self.repo.cache.local.cache_types = cache_types
+            self.repo.cache.legacy.cache_types = legacy_cache_types
 
     def _uncached_outs(self, stage, cache):
         # NOTE: using temporary stage to avoid accidentally modifying original
@@ -179,7 +182,7 @@ class StageCache:
         dump_yaml(tmp, cache)
         self.repo.cache.legacy.move(tmp, path)
 
-    def restore(self, stage, run_cache=True, pull=False, dry=False):
+    def restore(self, stage, run_cache=True, pull=False, dry=False):  # noqa: C901
         from .serialize import to_single_stage_lockfile
 
         if not _can_hash(stage):
@@ -203,8 +206,11 @@ class StageCache:
         cached_stage = self._create_stage(cache, wdir=stage.wdir)
 
         if pull and not dry:
-            for objs in cached_stage.get_used_objs().values():
-                self.repo.cloud.pull(objs)
+            try:
+                for objs in cached_stage.get_used_objs().values():
+                    self.repo.cloud.pull(objs)
+            except CollectCacheError as exc:
+                raise RunCacheNotFoundError(stage) from exc
 
         if not cached_stage.outs_cached():
             raise RunCacheNotFoundError(stage)

dvc/stage/run.py

@@ -80,7 +80,7 @@ def _run(executable, cmd, **kwargs):
     exec_cmd = _make_cmd(executable, cmd)
 
     try:
-        p = subprocess.Popen(exec_cmd, **kwargs)  # nosec B603
+        p = subprocess.Popen(exec_cmd, **kwargs)  # nosec B603 # noqa: S603
         if main_thread:
             old_handler = signal.signal(signal.SIGINT, signal.SIG_IGN)
 

dvc/stage/utils.py

@@ -32,14 +32,14 @@ def check_stage_path(repo, path, is_wdir=False):
         wdir_or_path="stage working dir" if is_wdir else "file path", path=path
     )
 
-    real_path = os.path.realpath(path)
+    real_path = os.path.abspath(path)
     if not os.path.exists(real_path):
         raise StagePathNotFoundError(error_msg.format("does not exist"))
 
     if not os.path.isdir(real_path):
         raise StagePathNotDirectoryError(error_msg.format("is not directory"))
 
-    proj_dir = os.path.realpath(repo.root_dir)
+    proj_dir = os.path.abspath(repo.root_dir)
     if real_path != proj_dir and not path_isin(real_path, proj_dir):
         raise StagePathOutsideError(error_msg.format("is outside of DVC repo"))
 
@@ -102,6 +102,10 @@ def check_no_externals(stage):
     link = format_link(
         "https://dvc.org/doc/user-guide/pipelines/external-dependencies-and-outputs"
     )
+    if stage.is_data_source:
+        link = format_link(
+            "https://dvc.org/doc/user-guide/data-management/importing-external-data"
+        )
     raise StageExternalOutputsError(
         f"Cached output(s) outside of DVC project: {str_outs}. "
         f"See {link} for more info."
@@ -280,3 +284,27 @@ def validate_kwargs(
         kwargs.pop("name", None)
 
     return kwargs
+
+
+def _get_stage_files(stage: "Stage") -> List[str]:
+    from dvc.dvcfile import ProjectFile
+    from dvc.utils import relpath
+
+    ret: List[str] = []
+    file = stage.dvcfile
+    ret.append(file.relpath)
+    if isinstance(file, ProjectFile):
+        ret.append(file._lockfile.relpath)  # pylint: disable=protected-access
+
+    for dep in stage.deps:
+        if (
+            not dep.use_scm_ignore
+            and dep.is_in_repo
+            and not stage.repo.dvcfs.isdvc(stage.repo.dvcfs.from_os_path(str(dep)))
+        ):
+            ret.append(relpath(dep.fs_path))
+
+    for out in stage.outs:
+        if not out.use_scm_ignore and out.is_in_repo:
+            ret.append(relpath(out.fs_path))
+    return ret

dvc/testing/benchmarks/conftest.py

@@ -1 +1 @@
-from .fixtures import *  # noqa, pylint: disable=wildcard-import
+from .fixtures import *  # noqa: F403 # pylint: disable=wildcard-import

dvc/testing/benchmarks/fixtures.py

@@ -90,7 +90,7 @@ def make_dvc_bin(
         dvc_bin = bench_config.dvc_bin
 
     def _dvc_bin(*args):
-        return check_output([dvc_bin, *args], text=True)  # nosec B603
+        return check_output([dvc_bin, *args], text=True)  # nosec B603  # noqa: S603
 
     _dvc_bin.version = parse_tuple(_dvc_bin("--version"))  # type: ignore[attr-defined]
     return _dvc_bin

dvc/testing/conftest.py

@@ -1 +1 @@
-from .fixtures import *  # noqa, pylint: disable=wildcard-import
+from .fixtures import *  # noqa: F403, pylint: disable=wildcard-import

dvc/testing/fixtures.py

@@ -233,8 +233,8 @@ def docker_services(
         pytest.skip("disabled for Windows on CI")
 
     try:
-        subprocess.check_output(  # nosec B607, B602
-            "docker ps", stderr=subprocess.STDOUT, shell=True
+        subprocess.check_output(  # nosec B607, B602,
+            "docker ps", stderr=subprocess.STDOUT, shell=True  # noqa: S602, S607
         )
     except subprocess.CalledProcessError as err:
         out = (err.output or b"").decode("utf-8")
@@ -242,7 +242,9 @@ def docker_services(
 
     try:
         cmd = "docker-compose version"
-        subprocess.check_output(cmd, stderr=subprocess.STDOUT, shell=True)  # nosec
+        subprocess.check_output(
+            cmd, stderr=subprocess.STDOUT, shell=True  # nosec B602 # noqa: S602
+        )
     except subprocess.CalledProcessError as err:
         out = (err.output or b"").decode("utf-8")
         pytest.skip(f"docker-compose is not installed: {out}")

dvc/testing/plugin.py

@@ -1,5 +1,5 @@
-from .benchmarks.fixtures import *  # noqa, pylint: disable=wildcard-import
-from .fixtures import *  # noqa, pylint: disable=wildcard-import
+from .benchmarks.fixtures import *  # noqa: F403, pylint: disable=wildcard-import
+from .fixtures import *  # noqa: F403, pylint: disable=wildcard-import
 
 
 def pytest_generate_tests(metafunc):

dvc/ui/__init__.py

@@ -147,7 +147,7 @@ class Console:
                 return self.write(json.text, stderr=stderr)
         return self.rich_print(json, stderr=stderr, soft_wrap=True)
 
-    def rich_print(
+    def rich_print(  # noqa: PLR0913
         self,
         *objects: Any,
         sep: str = " ",

dvc/ui/pager.py

@@ -59,7 +59,8 @@ def find_pager():
     if not pager:
         pager = os.getenv(PAGER_ENV)
     if not pager:
-        if os.system(f"({DEFAULT_PAGER}) 2>{os.devnull}") != 0:  # nosec B605
+        ret = os.system(f"({DEFAULT_PAGER}) 2>{os.devnull}")  # nosec B605 # noqa: S605
+        if ret != 0:
             logger.warning(
                 "Unable to find `less` in the PATH. Check out %s for more info.",
                 format_link("https://man.dvc.org/pipeline/show"),

dvc/utils/__init__.py

@@ -218,23 +218,13 @@ def _visual_center(line, width):
 
 
 def relpath(path, start=os.curdir):
-    path = os.fspath(path)
+    path = os.path.abspath(os.fspath(path))
     start = os.path.abspath(os.fspath(start))
 
     # Windows path on different drive than curdir doesn't have relpath
-    if os.name == "nt":
-        # Since python 3.8 os.realpath resolves network shares to their UNC
-        # path. So, to be certain that relative paths correctly captured,
-        # we need to resolve to UNC path first. We resolve only the drive
-        # name so that we don't follow any 'real' symlinks on the path
-        def resolve_network_drive_windows(path_to_resolve):
-            drive, tail = os.path.splitdrive(path_to_resolve)
-            return os.path.join(os.path.realpath(drive), tail)
+    if os.name == "nt" and not os.path.commonprefix([start, path]):
+        return path
 
-        path = resolve_network_drive_windows(os.path.abspath(path))
-        start = resolve_network_drive_windows(start)
-        if not os.path.commonprefix([start, path]):
-            return path
     return os.path.relpath(path, start)
 
 
@@ -319,7 +309,7 @@ def resolve_paths(repo, out, always_local=False):
 
 
 def format_link(link):
-    return "<{blue}{link}{nc}>".format(
+    return "<{blue}{link}{nc}>".format(  # noqa: UP032
         blue=colorama.Fore.CYAN, link=link, nc=colorama.Fore.RESET
     )
 

dvc/utils/serialize/__init__.py

@@ -1,11 +1,11 @@
 from collections import defaultdict
 from typing import DefaultDict
 
-from ._common import *  # noqa, pylint: disable=wildcard-import
-from ._json import *  # noqa, pylint: disable=wildcard-import
-from ._py import *  # noqa, pylint: disable=wildcard-import
-from ._toml import *  # noqa, pylint: disable=wildcard-import
-from ._yaml import *  # noqa, pylint: disable=wildcard-import
+from ._common import *  # noqa: F403, pylint: disable=wildcard-import
+from ._json import *  # noqa: F403, pylint: disable=wildcard-import
+from ._py import *  # noqa: F403, pylint: disable=wildcard-import
+from ._toml import *  # noqa: F403, pylint: disable=wildcard-import
+from ._yaml import *  # noqa: F403, pylint: disable=wildcard-import
 
 LOADERS: DefaultDict[str, LoaderFn] = defaultdict(  # noqa: F405
     lambda: load_yaml  # noqa: F405

dvc/utils/threadpool.py

@@ -46,7 +46,7 @@ class ThreadPoolExecutor(futures.ThreadPoolExecutor):
         if sys.version_info > (3, 9):  # pylint: disable=no-else-return
             # pylint: disable=unexpected-keyword-arg
             return super().shutdown(wait=wait, cancel_futures=cancel_futures)
-        else:
+        else:  # noqa: RET505
             with self._shutdown_lock:
                 self._shutdown = True
                 if cancel_futures:

pyproject.toml

@@ -34,13 +34,14 @@ dependencies = [
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=2.3.0,<2.4.0",
+    "dvc-data>=2.9.1,<2.10.0",
     "dvc-http>=2.29.0",
     "dvc-render>=0.3.1,<1",
     "dvc-studio-client>=0.9.2,<1",
     "dvc-task>=0.3.0,<1",
     "flatten_dict<1,>=0.4.1",
-    "flufl.lock>=5",
+    # https://github.com/iterative/dvc/issues/9654
+    "flufl.lock>=5,<8",
     "funcy>=1.14",
     "grandalf<1,>=0.7",
     "hydra-core>=1.1",
@@ -74,10 +75,11 @@ gdrive = ["dvc-gdrive==2.20"]
 gs = ["dvc-gs==2.22.1"]
 hdfs = ["dvc-hdfs==2.19"]
 lint = [
-    "mypy==1.3.0",
-    "pylint==2.17.4",
+    "mypy==1.4.1",
+    "pylint==2.17.5",
     "types-colorama",
     "types-psutil",
+    "types-pyinstaller",
     "types-requests",
     "types-tabulate",
     "types-toml",
@@ -128,6 +130,10 @@ dvc = "dvc.api:DVCFileSystem"
 [project.entry-points."pytest11"]
 dvc-testing = "dvc.testing.plugin"
 
+[project.entry-points."pyinstaller40"]
+hook-dirs = "dvc.__pyinstaller:get_hook_dirs"
+tests = "dvc.__pyinstaller:get_PyInstaller_tests"
+
 [tool.setuptools]
 license-files = ["LICENSE"]
 
@@ -262,8 +268,16 @@ ignored-argument-names = "_.*|args|kwargs"
 [tool.ruff]
 # external flake8 codes that should be preserved
 external = ["B301", "C901", "E302", "W601"]
-ignore = ["N818", "S101", "PT004", "PT007", "PT019", "SIM105", "SIM108", "SIM110", "SIM117", "TRY003", "TRY200", "TRY300", "TRY301", "PLR2004", "PLW2901", "RUF005"]
-select = ["F", "E", "W", "C90", "N", "UP", "YTT", "S", "BLE", "B", "A", "C4", "T10", "EXE", "ISC", "ICN", "G", "INP", "PIE", "T20", "PT", "Q", "RET501", "RET504", "SIM", "TID", "TCH", "ARG", "PGH", "PLC", "PLE", "PLR", "PLW", "TRY", "RUF"]
+ignore = [
+    "N818", "S101", "PT004", "PT007", "RET501", "RET502", "RET503", "SIM105", "SIM108",
+    "SIM117", "TRY003", "TRY200", "TRY300", "PLR2004", "PLW2901", "RUF012"
+]
+select = [
+    "F", "E", "W", "C90", "N", "UP", "YTT", "ASYNC", "S", "BLE", "B", "A", "C4", "T10",
+    "EXE", "ISC", "ICN", "G", "INP", "PIE", "T20", "PYI", "PT", "Q", "RSE", "RET",
+    "SLOT", "SIM", "TID", "TCH", "ARG", "PGH", "PLC", "PLE", "PLR", "PLW", "TRY",
+    "FLY", "PERF101", "RUF",
+]
 show-source = true
 
 [tool.ruff.flake8-pytest-style]
@@ -286,17 +300,17 @@ ignore-variadic-names = true
 known-first-party = ["dvc", "dvc_data", "dvc_objects", "dvc_render", "dvc_task", "tests"]
 
 [tool.ruff.pep8-naming]
-ignore-names = ["M", "SCM"]
+extend-ignore-names = ["M", "SCM"]
 
 [tool.ruff.per-file-ignores]
 "dvc/commands/**" = ["N806"]
 "dvc/testing/**" = ["ARG002"]
 "dvc/testing/benchmarks/**" = ["ARG001"]
-"scripts/**" = ["T201", "INP001", "TRY002"]
+"tests/**" = ["S", "ARG001", "ARG002", "TRY002", "TRY301"]
 
 [tool.ruff.pylint]
 max-args = 10
 
 [tool.bandit]
-exclude_dirs = ["tests", "scripts", "dvc/repo/experiments/utils.py"]
+exclude_dirs = ["tests", "dvc/repo/experiments/utils.py"]
 skips = ["B101"]

scripts/build-requirements.txt

@@ -1,3 +0,0 @@
-# NOTE: using strict version due to
-# https://github.com/iterative/dvc/issues/7949
-PyInstaller==5.9.0

scripts/build_package.sh

@@ -1,15 +0,0 @@
-#!/bin/bash
-
-set -e
-set -x
-
-if [ ! -d "dvc" ]; then
-  echo "Please run this script from repository root" >&2
-  exit 1
-fi
-
-echo 'PKG = "pip"' >dvc/_build.py
-
-python -m pip install build twine
-python -m build
-twine check dist/*

scripts/pyinstaller/.gitignore

@@ -1,4 +0,0 @@
-/build
-/dist
-/dvc.spec
-/dvc

scripts/pyinstaller/build.py

@@ -1,47 +0,0 @@
-import os
-import pathlib
-from subprocess import STDOUT, check_call, check_output
-
-path = pathlib.Path(__file__).parent.absolute()
-hooks = path / "hooks"
-dvc = path.parent.parent / "dvc"
-entry = dvc / "__main__.py"
-
-check_call(
-    [
-        "pyinstaller",
-        "--additional-hooks-dir",
-        os.fspath(hooks),
-        "--name",
-        "dvc",
-        "-y",
-        os.fspath(entry),
-    ],
-    cwd=path,
-    stderr=STDOUT,
-)
-
-out = check_output(
-    [
-        path / "dist" / "dvc" / "dvc",
-        "doctor",
-    ],
-    stderr=STDOUT,
-).decode()
-
-remotes = [
-    "s3",
-    "oss",
-    "gdrive",
-    "gs",
-    "hdfs",
-    "http",
-    "webhdfs",
-    "azure",
-    "ssh",
-    "webdav",
-]
-
-print(out)
-for remote in remotes:
-    assert f"\t{remote}" in out, f"Missing support for {remote}"

scripts/pyinstaller/hooks/hook-dvc.system.py

@@ -1,4 +0,0 @@
-import os
-
-if os.name == "nt":
-    hiddenimports = ["win32file"]

scripts/pyinstaller/hooks/hook-dvc.tree.gs.py

@@ -1,9 +0,0 @@
-from PyInstaller.utils.hooks import copy_metadata  # pylint:disable=import-error
-
-datas = copy_metadata("google-cloud-storage")
-
-# NOTE: including "requests" metadata because of google-resumable-media
-# bug. See [1] for more info.
-#
-# [1] https://github.com/googleapis/google-resumable-media-python/issues/59
-datas += copy_metadata("requests")

scripts/pyinstaller/hooks/hook-google_compute_engine.logger.py

@@ -1,4 +0,0 @@
-from PyInstaller.utils.hooks import copy_metadata  # pylint:disable=import-error
-
-datas = copy_metadata("google-compute-engine")
-hiddenimports = ["google_cloud_engine"]

tests/conftest.py

@@ -7,11 +7,11 @@ import pytest
 
 from dvc import env
 from dvc.stage import PipelineStage
-from dvc.testing.fixtures import *  # noqa, pylint: disable=wildcard-import
+from dvc.testing.fixtures import *  # noqa: F403, pylint: disable=wildcard-import
 
-from .dir_helpers import *  # noqa, pylint: disable=wildcard-import
-from .remotes import *  # noqa, pylint: disable=wildcard-import
-from .scripts import *  # noqa, pylint: disable=wildcard-import
+from .dir_helpers import *  # noqa: F403 # pylint: disable=wildcard-import
+from .remotes import *  # noqa: F403 # pylint: disable=wildcard-import
+from .scripts import *  # noqa: F403, pylint: disable=wildcard-import
 
 # Prevent updater and analytics from running their processes
 os.environ["DVC_TEST"] = "true"

tests/func/api/test_data.py

@@ -5,7 +5,7 @@ from funcy import first, get_in
 
 from dvc import api
 from dvc.exceptions import OutputNotFoundError, PathMissingError
-from dvc.testing.api_tests import TestAPI  # noqa, pylint: disable=unused-import
+from dvc.testing.api_tests import TestAPI  # noqa: F401, pylint: disable=unused-import
 from dvc.testing.tmp_dir import make_subrepo
 from dvc.utils.fs import remove
 
@@ -198,11 +198,13 @@ def test_get_url_subrepos(tmp_dir, scm, local_cloud):
         local_cloud / "files" / "md5" / "ac" / "bd18db4cc2f85cedef654fccc4a4d8"
     )
     assert api.get_url(os.path.join("subrepo", "dir", "foo")) == expected_url
+    assert api.get_url(os.path.join("subrepo", "dir", "foo"), repo=".") == expected_url
 
     expected_url = os.fspath(
         local_cloud / "files" / "md5" / "37" / "b51d194a7513e45b56f6524f2d51f2"
     )
     assert api.get_url("subrepo/bar") == expected_url
+    assert api.get_url("subrepo/bar", repo=".") == expected_url
 
 
 def test_open_from_remote(tmp_dir, erepo_dir, cloud, local_cloud):

tests/func/api/test_show.py

@@ -187,7 +187,7 @@ def test_params_show_stages(params_repo):
 
 
 def test_params_show_stage_addressing(tmp_dir, dvc):
-    for subdir in {"subdir1", "subdir2"}:
+    for subdir in ("subdir1", "subdir2"):
         subdir = tmp_dir / subdir
         subdir.mkdir()
         with subdir.chdir():
@@ -195,7 +195,7 @@ def test_params_show_stage_addressing(tmp_dir, dvc):
 
             dvc.run(name="stage-0", cmd="echo stage-0", params=["foo"])
 
-    for s in {"subdir1", "subdir2"}:
+    for s in ("subdir1", "subdir2"):
         dvcyaml = os.path.join(s, "dvc.yaml")
         assert api.params_show(stages=f"{dvcyaml}:stage-0") == {"foo": 1}
 

tests/func/artifacts/test_artifacts.py

@@ -1,4 +1,6 @@
+import logging
 import os
+from copy import deepcopy
 
 import pytest
 
@@ -39,6 +41,22 @@ def test_artifacts_read_subdir(tmp_dir, dvc):
     }
 
 
+def test_artifacts_read_bad_name(tmp_dir, dvc, caplog):
+    bad_name_dvcyaml = deepcopy(dvcyaml)
+    bad_name_dvcyaml["artifacts"]["bad_name"] = {"type": "model", "path": "bad.pkl"}
+
+    (tmp_dir / "dvc.yaml").dump(bad_name_dvcyaml)
+
+    artifacts = {
+        name: Artifact(**values)
+        for name, values in bad_name_dvcyaml["artifacts"].items()
+    }
+
+    with caplog.at_level(logging.WARNING):
+        assert tmp_dir.dvc.artifacts.read() == {"dvc.yaml": artifacts}
+        assert "Can't use 'bad_name' as artifact name (ID)" in caplog.text
+
+
 def test_artifacts_add_subdir(tmp_dir, dvc):
     subdir = tmp_dir / "subdir"
     subdir.mkdir()

tests/func/experiments/conftest.py

@@ -1,6 +1,7 @@
 import pytest
 
-from tests.unit.repo.experiments.conftest import (  # noqa, pylint disable=unused-argument
+# pylint: disable-next=unused-argument
+from tests.unit.repo.experiments.conftest import (  # noqa: F401
     exp_stage,
     failed_exp_stage,
     session_app,

tests/func/experiments/executor/test_ssh.py

@@ -10,7 +10,9 @@ from dvc_ssh.tests.cloud import TEST_SSH_KEY_PATH, TEST_SSH_USER
 from dvc.repo.experiments.executor.base import ExecutorInfo, ExecutorResult
 from dvc.repo.experiments.executor.ssh import SSHExecutor
 from dvc.repo.experiments.refs import EXEC_HEAD, EXEC_MERGE
-from tests.func.machine.conftest import *  # noqa, pylint: disable=wildcard-import
+
+# pylint: disable-next=wildcard-import
+from tests.func.machine.conftest import *  # noqa: F403
 
 
 @contextmanager
@@ -119,7 +121,7 @@ def test_reproduce(tmp_dir, scm, dvc, cloud, exp_stage, mocker):
         rev,
         fs_factory=partial(_ssh_factory, cloud),
     )
-    assert mock_execute.called_once()
+    mock_execute.assert_called_once()
     _name, args, _kwargs = mock_execute.mock_calls[0]
     assert f"dvc exp exec-run --infofile {infofile}" in args[0]
 
@@ -148,7 +150,7 @@ def test_run_machine(tmp_dir, scm, dvc, cloud, exp_stage, mocker):
 
     tmp_dir.gen("params.yaml", "foo: 2")
     dvc.experiments.run(exp_stage.addressing, machine="foo")
-    assert mock_repro.called_once()
+    mock_repro.assert_called_once()
     _name, _args, kwargs = mock_repro.mock_calls[0]
     info = kwargs["info"]
     url = urlparse(info.git_url)

tests/func/experiments/test_experiments.py

@@ -488,16 +488,6 @@ def test_run_celery(tmp_dir, scm, dvc, exp_stage, mocker):
     assert expected == metrics
 
 
-def test_run_metrics(tmp_dir, scm, dvc, exp_stage, mocker):
-    from dvc.cli import main
-
-    mocker.patch.object(dvc.experiments, "run", return_value={"abc123": "abc123"})
-    show_mock = mocker.patch.object(dvc.metrics, "show", return_value={})
-
-    main(["exp", "run", "-m"])
-    assert show_mock.called_once()
-
-
 def test_checkout_targets_deps(tmp_dir, scm, dvc, exp_stage):
     from dvc.utils.fs import remove
 

tests/func/experiments/test_show.py

@@ -452,7 +452,7 @@ def test_show_csv(tmp_dir, scm, dvc, exp_stage, capsys):
     assert "metrics.yaml:foo,params.yaml:foo,copy.py" in cap.out
     assert f",workspace,baseline,,,3,3,{data_hash}" in cap.out
     assert (
-        ",master,baseline,{},,1,1,{}".format(
+        ",master,baseline,{},,1,1,{}".format(  # noqa: UP032
             _get_rev_isotimestamp(baseline_rev), data_hash
         )
         in cap.out
@@ -682,7 +682,7 @@ def test_metrics_renaming(tmp_dir, dvc, scm, capsys, copy_script):
 
     assert f",master,baseline,{_get_rev_isotimestamp(scores_rev)},,1,,1" in cap.out
     assert (
-        ",{},baseline,{},,,1,1".format(
+        ",{},baseline,{},,,1,1".format(  # noqa: UP032
             metrics_rev[:7], _get_rev_isotimestamp(metrics_rev)
         )
         in cap.out

tests/func/repro/test_repro.py

@@ -10,7 +10,7 @@ from funcy import lsplit
 from dvc.cli import main
 from dvc.dvcfile import LOCK_FILE, PROJECT_FILE
 from dvc.exceptions import CyclicGraphError, ReproductionError
-from dvc.fs import LocalFileSystem, system
+from dvc.fs import system
 from dvc.output import Output
 from dvc.stage import PipelineStage, Stage
 from dvc.stage.cache import RunCacheNotSupported
@@ -624,12 +624,14 @@ class TestReproAlreadyCached:
         assert stage.outs[0].hash_info != saved_stage.outs[0].hash_info
 
     def test_force_import(self, mocker, tmp_dir, dvc):
+        from dvc.dependency import base
+
         tmp_dir.dvc_gen("foo", "foo")
 
         ret = main(["import-url", "foo", "bar"])
         assert ret == 0
 
-        spy_get = mocker.spy(LocalFileSystem, "get")
+        spy_get = mocker.spy(base, "fs_download")
         spy_checkout = mocker.spy(Output, "checkout")
 
         assert main(["unfreeze", "bar.dvc"]) == 0
@@ -848,9 +850,7 @@ def test_repro_dry_no_exec(tmp_dir, dvc):
                 idir,
                 "-o",
                 odir,
-                'python -c \'import shutil; shutil.copytree("{}", "{}")\''.format(
-                    idir, odir
-                ),
+                f'python -c \'import shutil; shutil.copytree("{idir}", "{odir}")\'',
             ]
         )
         assert ret == 0
@@ -943,6 +943,18 @@ def test_repro_force_downstream(tmp_dir, dvc, copy_script):
     assert stages[1].addressing == file3_stage.addressing
 
 
+def test_repro_force_downstream_do_not_force_independent_stages(tmp_dir, dvc, run_copy):
+    tmp_dir.gen({"foo": "foo", "bar": "bar"})
+    foo1 = run_copy("foo", "foo1", name="foo1")
+    foo2 = run_copy("foo1", "foo2", name="foo2")
+    run_copy("bar", "bar1", name="bar1")
+    run_copy("bar1", "bar2", name="bar2")
+    cat = dvc.run(cmd="cat bar2 foo2", deps=["foo2", "bar2"], name="cat")
+
+    tmp_dir.gen("foo", "foobar")
+    assert dvc.reproduce(force_downstream=True) == [foo1, foo2, cat]
+
+
 def test_repro_pipeline(
     tmp_dir,
     dvc,
@@ -1282,3 +1294,67 @@ def test_repro_rm_recursive(tmp_dir, dvc):
     dvc.reproduce()
     assert (tmp_dir / "dir").exists()
     assert not (tmp_dir / "dir" / "foo").exists()
+
+
+def test_repro_single_item_with_multiple_targets(tmp_dir, dvc, copy_script):
+    stage1 = dvc.stage.add(cmd="echo foo > foo", outs=["foo"], name="gen-foo")
+    with dvc.lock:
+        stage1.run()
+
+    stage2 = dvc.stage.add(
+        cmd="python copy.py foo bar", deps=["foo"], outs=["bar"], name="copy-foo-bar"
+    )
+    assert dvc.reproduce(["copy-foo-bar", "gen-foo"], single_item=True) == [
+        stage2,
+        stage1,
+    ]
+
+
+def test_repro_keep_going(mocker, tmp_dir, dvc, copy_script):
+    from dvc.repo import reproduce
+
+    (bar_stage, foo_stage) = tmp_dir.dvc_gen({"bar": "bar", "foo": "foo"})
+    stage1 = dvc.stage.add(
+        cmd=["python copy.py bar foobar", "exit 1"],
+        deps=["bar"],
+        outs=["foobar"],
+        name="copy-bar-foobar",
+    )
+    dvc.stage.add(cmd="cat foobar foo", deps=["foobar", "foo"], name="cat")
+    spy = mocker.spy(reproduce, "_reproduce_stage")
+
+    with pytest.raises(ReproductionError):
+        dvc.reproduce(on_error="keep-going", repro_fn=spy)
+
+    assert spy.call_args_list == [
+        mocker.call(bar_stage, upstream=[], force=False, interactive=False),
+        mocker.call(stage1, upstream=[bar_stage], force=False, interactive=False),
+        mocker.call(foo_stage, upstream=[], force=False, interactive=False),
+    ]
+
+
+def test_repro_ignore_errors(M, mocker, tmp_dir, dvc, copy_script):
+    from dvc.repo import reproduce
+
+    (bar_stage, foo_stage) = tmp_dir.dvc_gen({"bar": "bar", "foo": "foo"})
+    stage1 = dvc.stage.add(
+        cmd=["python copy.py bar foobar", "exit 1"],
+        deps=["bar"],
+        outs=["foobar"],
+        name="copy-bar-foobar",
+    )
+    stage2 = dvc.stage.add(cmd="cat foobar foo", deps=["foobar", "foo"], name="cat")
+    spy = mocker.spy(reproduce, "_reproduce_stage")
+    dvc.reproduce(on_error="ignore", repro_fn=spy)
+
+    assert spy.call_args_list == [
+        mocker.call(bar_stage, upstream=[], force=False, interactive=False),
+        mocker.call(stage1, upstream=[bar_stage], force=False, interactive=False),
+        mocker.call(foo_stage, upstream=[], force=False, interactive=False),
+        mocker.call(
+            stage2,
+            upstream=[foo_stage, stage1],
+            force=False,
+            interactive=False,
+        ),
+    ]