(fix) DVC_SITE_CACHE_DIR environment variable should override core.site_cache_dir (#10778)

* Prioritize DVC_SITE_CACHE_DIR over core.site_cache_dir

* Add tests for proper prioritization of env var and config

* Tests require python-3.9 compatible typing, use Optional

* Remove comment

* Fixup

.github/ISSUE_TEMPLATE/epic_story.md

@@ -1,60 +0,0 @@
----
-name: Epic/Story Issue Template
-about: Template for "top" level issues - Epics (>2 weeks) / Stories (<2 weeks)
-title: 'Epic: New Feature'
-labels: epic
-assignees:
-
----
-
-## Summary / Background
-What do you want to achieve, why? business context
-...
-
-## Scope
-
-What will be impacted and what won't be?
-What needs implementation and what is invariant?
-e.g.
-- user should be able to run workflow X from UI
-- enable workflow Y, Z from CLI
-
-## Assumptions
-Product / UX assumptions as well as technical assumptions / limitations
-e.g.
-* Support only Python Runtime
-* Focus on DVC experiments only
-* Deployment environments don't change often and can be picked up from shared configuration
-
-## Open Questions
-e.g.
-- How should access control work for shared artifacts (workflow X)
-- Python runtime assumption - is it really valid? in light of <...>
-
-## Blockers / Dependencies
-List issues or other conditions / blockers
-
-## General Approach
-Invocation example:
-```shell
-$ mapper-run task.tar.gz --ray-cluster <ip>:<port>
-```
-
-## Steps
-
-### Must have (p1)
-- [ ] subissue2
-- [ ] step 2
- - info
- - info
-
-### Optional / followup (p2)
-- [ ] ⌛ step 3 wip
-- [ ] step 4
-
-## Timelines
-
-Put your estimations here. Update once certainty changes
-- end of week (Feb 3) for prototype with workflows X, Y
-- Feb 15 - MVP in production
-- Low priority followups can be done later

.github/workflows/benchmarks.yaml

@@ -5,6 +5,10 @@ concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
   cancel-in-progress: true
 
+env:
+  FORCE_COLOR: "1"
+  PY_COLORS: "1"
+
 jobs:
   bench:
     name: run benchmarks
@@ -12,7 +16,29 @@ jobs:
     steps:
       - uses: actions/setup-python@v5
         with:
-            python-version: "3.11"
-      - uses: iterative/dvc-bench@main
+            python-version: "3.13"
+
+      - uses: actions/checkout@v4
         with:
-            pytest_options: "-k 'test_init or test_help'"
+          ref: ${{ github.event.pull_request.base.sha }}
+          fetch-depth: 0
+
+      - uses: astral-sh/setup-uv@v6
+        with:
+          enable-cache: true
+          cache-dependency-glob: pyproject.toml
+
+      - run: uv pip install '.[tests]' --system
+      - name: run benchmarks on base branch
+        run: pytest --benchmark-autosave dvc/testing/benchmarks/ -k 'test_init or test_help'
+
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+          clean: false
+      - run: uv pip install '.[tests]' --system
+      - name: run benchmarks for PR
+        run: >
+          pytest --benchmark-compare --benchmark-compare-fail=min:5%
+          --benchmark-group-by name
+          dvc/testing/benchmarks/ -k 'test_init or test_help'

.github/workflows/build.yaml

@@ -25,24 +25,27 @@ jobs:
 
     - uses: actions/setup-python@v5
       with:
-        python-version: 3.12
+        python-version: 3.13
 
-    - run: python -m pip install --upgrade pip setuptools_scm build twine
+    - uses: astral-sh/setup-uv@v6
 
-    - name: Force version for Test PyPI upload
-      if: ${{ !startsWith(github.ref, 'refs/tags') }}
-      run: echo version=$(python -m setuptools_scm | awk -F+ '{print $1}' | tail -1) >> $GITHUB_ENV
+    - name: Omit local version for Test PyPI upload
+      if: github.event_name == 'push' && github.ref == 'refs/heads/main'
+      run: echo SETUPTOOLS_SCM_OVERRIDES_FOR_DVC='{local_scheme="no-local-version"}' >> $GITHUB_ENV
 
-    - run: echo 'PKG = "pip"'>dvc/_build.py
-    - run: python -m build
-      env:
-        SETUPTOOLS_SCM_PRETEND_VERSION_FOR_DVC: ${{ env.version }}
-    - run: twine check --strict dist/*
+    - name: Build Python Package
+      run: |
+        echo 'PKG = "pip"'>dvc/_build.py
+        uv build
+
+    - name: Check dist
+      run: uv tool run twine check --strict dist/*
 
     - uses: actions/upload-artifact@v4
       with:
         name: Packages
         path: dist/
+        if-no-files-found: error
 
   test-pypi-publish:
     name: Publish dev package to test.pypi.org
@@ -52,7 +55,7 @@ jobs:
 
     environment:
       name: test-pypi
-      url: https://test.pypi.org/p/dvc
+      url: https://test.pypi.org/p/dvc/${{ github.event.release.tag_name }}
 
     permissions:
       id-token: write
@@ -77,7 +80,7 @@ jobs:
 
     environment:
       name: pypi
-      url: https://pypi.org/p/dvc
+      url: https://pypi.org/p/dvc/${{ github.event.release.tag_name }}
 
     permissions:
       id-token: write

.github/workflows/plugin_tests.yaml

@@ -18,8 +18,8 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-20.04]
-        pyv: ["3.10"]
+        os: [ubuntu-latest]
+        pyv: ["3.13"]
         plugin: ["dvc-s3"]
 
     steps:
@@ -37,16 +37,17 @@ jobs:
       uses: actions/setup-python@v5
       with:
         python-version: ${{ matrix.pyv }}
-        cache: pip
-        cache-dependency-path: |
-          dvc/pyproject.toml
-          ${{ matrix.plugin }}/setup.cfg
+
+    - uses: astral-sh/setup-uv@v6
+      with:
+        enable-cache: true
+        cache-dependency-glob: "**/pyproject.toml"
+        cache-suffix: ${{ matrix.pyv }}
 
     - name: Install plugin + DVC@PR
       run: |
-        pip install --upgrade pip wheel
-        pip install "./dvc[testing]"
-        pip install -e "./${{ matrix.plugin }}[tests]"
+        uv pip install "./dvc[testing]" --system
+        uv pip install -e "./${{ matrix.plugin }}[tests]" --system
 
     - name: Run plugin tests
       timeout-minutes: 15

.github/workflows/tests.yaml

@@ -16,15 +16,107 @@ permissions:
   contents: read
 
 jobs:
+  lint:
+    timeout-minutes: 10
+    runs-on: ${{ matrix.os }}
+    env:
+       DVC_NO_ANALYTICS: true
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-latest, macos-latest, windows-latest]
+    steps:
+    - uses: actions/checkout@v4
+      with:
+        fetch-depth: 0
+    - name: Set up Python
+      uses: actions/setup-python@v5
+      with:
+        python-version: '3.13'
+    - uses: astral-sh/setup-uv@v6
+      with:
+        enable-cache: true
+        cache-dependency-glob: pyproject.toml
+    - run: uv pip install -e ".[dev]" --system
+      env:
+        UV_LINK_MODE: ${{ runner.os == 'Windows' && 'symlink' || 'hardlink' }}
+
+    - name: Cache mypy
+      uses: actions/cache@v4
+      with:
+        path: .mypy_cache
+        key: mypy-${{ runner.os }}-${{ env.pythonLocation }}-${{ hashFiles('pyproject.toml') }}
+    - uses: pre-commit/[email protected]
+
   tests:
-    timeout-minutes: 50
+    timeout-minutes: 20
     runs-on: ${{ matrix.os }}
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-20.04, windows-latest, macos-latest]
-        pyv: ["3.9", "3.10", "3.11", "3.12"]
-
+        os: [ubuntu-latest, macos-latest]
+        pyv: ["3.9", "3.10", "3.11", "3.12", "3.13"]
+        include:
+        - os: windows-latest
+          pyv: "3.9"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 1"
+        - os: windows-latest
+          pyv: "3.9"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 2"
+        - os: windows-latest
+          pyv: "3.9"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 3"
+        - os: windows-latest
+          pyv: "3.9"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 4"
+        - os: windows-latest
+          pyv: "3.10"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 1"
+        - os: windows-latest
+          pyv: "3.10"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 2"
+        - os: windows-latest
+          pyv: "3.10"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 3"
+        - os: windows-latest
+          pyv: "3.10"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 4"
+        - os: windows-latest
+          pyv: "3.11"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 1"
+        - os: windows-latest
+          pyv: "3.11"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 2"
+        - os: windows-latest
+          pyv: "3.11"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 3"
+        - os: windows-latest
+          pyv: "3.11"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 4"
+        - os: windows-latest
+          pyv: "3.12"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 1"
+        - os: windows-latest
+          pyv: "3.12"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 2"
+        - os: windows-latest
+          pyv: "3.12"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 3"
+        - os: windows-latest
+          pyv: "3.12"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 4"
+        - os: windows-latest
+          pyv: "3.13"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 1"
+        - os: windows-latest
+          pyv: "3.13"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 2"
+        - os: windows-latest
+          pyv: "3.13"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 3"
+        - os: windows-latest
+          pyv: "3.13"
+          pytestargs: "--splitting-algorithm=least_duration --splits 4 --group 4"
     steps:
     - uses: actions/checkout@v4
       with:
@@ -33,25 +125,24 @@ jobs:
       uses: actions/setup-python@v5
       with:
         python-version: ${{ matrix.pyv }}
-        cache: pip
-        cache-dependency-path: pyproject.toml
-    - name: install
-      run: |
-        pip install --upgrade pip wheel
-        pip install -e ".[dev]"
-    - uses: pre-commit/[email protected]
+    - uses: astral-sh/setup-uv@v6
+      with:
+        enable-cache: true
+        cache-dependency-glob: pyproject.toml
+        cache-suffix: ${{ matrix.pyv }}
+    - run: uv pip install -e ".[dev]" pytest-split --system
+      env:
+        UV_LINK_MODE: ${{ runner.os == 'Windows' && 'symlink' || 'hardlink' }}
+
     - name: run tests
       timeout-minutes: 40
       env:
         PYTHONUTF8: 1
       run: >
-        pytest -n=logical --timeout=300 --durations=100
-        --cov --cov-report=xml --cov-report=term
+        pytest ${{ matrix.pytestargs }} -n=logical --dist=worksteal --timeout=300 --durations=0
+        --cov --cov-report=xml --cov-report=term  --durations-path=./.github/.test_durations
     - name: upload coverage report
-      uses: codecov/codecov-action@v3
-      with:
-        file: ./coverage.xml
-        fail_ci_if_error: false
+      uses: codecov/codecov-action@v5
   check:
     if: always()
     needs: [tests]
@@ -66,7 +157,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: Slack Notification
-      uses: rtCamp/[email protected].0
+      uses: rtCamp/[email protected].3
       env:
         SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}
         SLACK_COLOR: ${{ job.status }}

.gitignore

@@ -43,7 +43,7 @@ scripts/ci/gcp-creds.json
 .gcp-creds.json
 
 *~
-/dvc/_dvc_version.py
+/dvc/_version.py
 
 .mypy_cache/
 .pytest_cache/

.pre-commit-config.yaml

@@ -6,7 +6,7 @@ ci:
 
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.5.0
+    rev: v5.0.0
     hooks:
       - id: check-added-large-files
       - id: check-case-conflict
@@ -22,13 +22,13 @@ repos:
       - id: sort-simple-yaml
       - id: trailing-whitespace
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.3.4'
+    rev: 'v0.11.13'
     hooks:
-      - id: ruff
+      - id: ruff-check
         args: [--fix, --exit-non-zero-on-fix]
       - id: ruff-format
   - repo: https://github.com/codespell-project/codespell
-    rev: v2.2.6
+    rev: v2.4.1
     hooks:
       - id: codespell
         additional_dependencies: ["tomli"]
@@ -50,7 +50,7 @@ repos:
         language: system
         name: DVC pre-commit
         stages:
-          - commit
+          - pre-commit
         verbose: true
         require_serial: true
       - args:
@@ -61,7 +61,7 @@ repos:
         language: system
         name: DVC pre-push
         stages:
-          - push
+          - pre-push
         require_serial: true
       - always_run: true
         args:

.pre-commit-hooks.yaml

@@ -8,7 +8,7 @@
   name: DVC pre-commit
   require_serial: true
   stages:
-  - commit
+  - pre-commit
   verbose: true
 - args:
   - git-hook
@@ -20,7 +20,7 @@
   name: DVC pre-push
   require_serial: true
   stages:
-  - push
+  - pre-push
 - always_run: true
   args:
   - git-hook
@@ -29,7 +29,7 @@
   id: dvc-post-checkout
   language: python
   language_version: python3
-  minimum_pre_commit_version: 2.2.0
+  minimum_pre_commit_version: 3.2.0
   name: DVC post-checkout
   require_serial: true
   stages:

dvc/api/__init__.py

@@ -1,7 +1,7 @@
 from dvc.fs.dvc import _DVCFileSystem as DVCFileSystem
 
 from .artifacts import artifacts_show
-from .data import get_url, open, read
+from .data import get_url, open, read  # noqa: A004
 from .dataset import get as get_dataset
 from .experiments import exp_save, exp_show
 from .scm import all_branches, all_commits, all_tags

dvc/api/artifacts.py

@@ -1,3 +1,4 @@
+import os
 from typing import Any, Optional
 
 from dvc.repo import Repo
@@ -36,12 +37,22 @@ def artifacts_show(
     if version and stage:
         raise ValueError("Artifact version and stage are mutually exclusive.")
 
+    from dvc.repo.artifacts import Artifacts
+    from dvc.utils import as_posix
+
     repo_kwargs: dict[str, Any] = {
         "subrepos": True,
         "uninitialized": True,
     }
+
+    dirname, _ = Artifacts.parse_path(name)
     with Repo.open(repo, **repo_kwargs) as _repo:
         rev = _repo.artifacts.get_rev(name, version=version, stage=stage)
         with _repo.switch(rev):
-            path = _repo.artifacts.get_path(name)
-        return {"rev": rev, "path": path}
+            root = _repo.fs.root_marker
+            _dirname = _repo.fs.join(root, dirname) if dirname else root
+            with Repo(_dirname, fs=_repo.fs, scm=_repo.scm) as r:
+                path = r.artifacts.get_path(name)
+                path = _repo.fs.join(_repo.fs.root_marker, as_posix(path))
+                parts = _repo.fs.relparts(path, _repo.root_dir)
+                return {"rev": rev, "path": os.path.join(*parts)}

dvc/api/data.py

@@ -285,7 +285,7 @@ def _open(
             if TYPE_CHECKING:
                 from dvc.fs import FileSystem
 
-            fs: Union["FileSystem", DataFileSystem, DVCFileSystem]
+            fs: Union[FileSystem, DataFileSystem, DVCFileSystem]
             if os.path.isabs(path):
                 fs = DataFileSystem(index=_repo.index.data["local"])
                 fs_path = path

dvc/api/dataset.py

@@ -1,8 +1,8 @@
 from typing import Literal, TypedDict, Union
 
 
-class DVCXDataset(TypedDict):
-    type: Literal["dvcx"]
+class DatachainDataset(TypedDict):
+    type: Literal["dc"]
     name: str
     version: int
 
@@ -20,7 +20,7 @@ class URLDataset(TypedDict):
     path: str
 
 
-def get(name: str) -> Union[DVCXDataset, DVCDataset, URLDataset]:
+def get(name: str) -> Union[DatachainDataset, DVCDataset, URLDataset]:
     from difflib import get_close_matches
 
     from dvc.fs import get_cloud_fs
@@ -46,9 +46,9 @@ def get(name: str) -> Union[DVCXDataset, DVCDataset, URLDataset]:
             path=dataset.lock.path,
             sha=dataset.lock.rev_lock,
         )
-    if dataset.type == "dvcx":
-        return DVCXDataset(
-            type="dvcx", name=dataset.name_version[0], version=dataset.lock.version
+    if dataset.type == "dc":
+        return DatachainDataset(
+            type="dc", name=dataset.name_version[0], version=dataset.lock.version
         )
     if dataset.type == "url":
         fs_cls, _, path = get_cloud_fs(repo.config, url=dataset.lock.url)

dvc/api/show.py

@@ -27,8 +27,7 @@ def _postprocess(results):
             }
             processed[rev] = processed[rev] | to_merge
 
-    if "workspace" in processed:
-        del processed["workspace"]
+    processed.pop("workspace", None)
 
     return processed
 

dvc/cli/command.py

@@ -18,8 +18,8 @@ class CmdBase(ABC):
 
         os.chdir(args.cd)
 
-        self.repo: "Repo" = Repo(uninitialized=self.UNINITIALIZED)
-        self.config: "Config" = self.repo.config
+        self.repo: Repo = Repo(uninitialized=self.UNINITIALIZED)
+        self.config: Config = self.repo.config
         self.args = args
 
     def do_run(self):

dvc/cli/utils.py

@@ -40,7 +40,7 @@ def hide_subparsers_from_help(subparsers):
     # from the 'positional arguments' choices list
     # see: https://bugs.python.org/issue22848
     # Need to set `add_help=False`, but avoid setting `help`
-    # (not even to `argparse.SUPPPRESS`).
+    # (not even to `argparse.SUPPRESS`).
     # NOTE: The argument is the parent subparser, not the subcommand parser.
     cmds = [cmd for cmd, parser in subparsers.choices.items() if parser.add_help]
     subparsers.metavar = "{{{}}}".format(",".join(cmds))

dvc/commands/add.py

@@ -51,6 +51,7 @@ class CmdAdd(CmdBase):
                 to_remote=self.args.to_remote,
                 remote_jobs=self.args.remote_jobs,
                 force=self.args.force,
+                relink=self.args.relink,
             )
         except FileNotFoundError:
             logger.exception("")
@@ -119,6 +120,13 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Override local file or folder if exists.",
     )
+    parser.add_argument(
+        "--no-relink",
+        dest="relink",
+        action="store_false",
+        help="Don't recreate links from cache to workspace.",
+    )
+    parser.set_defaults(relink=True)
     parser.add_argument(
         "targets", nargs="+", help="Input files/directories to add."
     ).complete = completion.FILE

dvc/commands/artifacts.py

@@ -13,6 +13,9 @@ class CmdArtifactsGet(CmdBaseNoRepo):
         from dvc.scm import CloneError
         from dvc.ui import ui
 
+        if self.args.show_url:
+            return self._show_url()
+
         try:
             count, out = Artifacts.get(
                 self.args.url,
@@ -36,6 +39,28 @@ class CmdArtifactsGet(CmdBaseNoRepo):
             )
             return 1
 
+    def _show_url(self):
+        from dvc.api import artifacts_show, get_url
+        from dvc.ui import ui
+
+        artifact = artifacts_show(
+            self.args.name,
+            version=self.args.rev,
+            stage=self.args.stage,
+            repo=self.args.url,
+        )
+
+        url = get_url(
+            artifact["path"],
+            repo=self.args.url,
+            rev=artifact["rev"],
+            remote=self.args.remote,
+            remote_config=self.args.remote_config,
+        )
+        ui.write(url, force=True)
+
+        return 0
+
 
 def add_parser(subparsers, parent_parser):
     ARTIFACTS_HELP = "DVC model registry artifact commands."
@@ -84,6 +109,14 @@ def add_parser(subparsers, parent_parser):
         help="Destination path to download artifact to",
         metavar="<path>",
     ).complete = completion.DIR
+    get_parser.add_argument(
+        "--show-url",
+        action="store_true",
+        help=(
+            "Print the storage location (URL) the target data would be "
+            "downloaded from, and exit."
+        ),
+    )
     get_parser.add_argument(
         "-j",
         "--jobs",

dvc/commands/commit.py

@@ -20,6 +20,7 @@ class CmdCommit(CmdBase):
                     with_deps=self.args.with_deps,
                     recursive=self.args.recursive,
                     force=self.args.force,
+                    relink=self.args.relink,
                 )
             except DvcException:
                 logger.exception("failed to commit%s", (" " + target) if target else "")
@@ -64,6 +65,13 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Commit cache for subdirectories of the specified directory.",
     )
+    commit_parser.add_argument(
+        "--no-relink",
+        dest="relink",
+        action="store_false",
+        help="Don't recreate links from cache to workspace.",
+    )
+    commit_parser.set_defaults(relink=True)
     commit_parser.add_argument(
         "targets",
         nargs="*",

dvc/commands/dag.py

@@ -17,7 +17,7 @@ def _show_ascii(graph: "DiGraph"):
 
     ret = []
     for pipeline in pipelines:
-        ret.append(draw(pipeline.nodes, pipeline.edges))
+        ret.append(draw(pipeline.nodes, pipeline.edges))  # noqa: PERF401
 
     return "\n".join(ret)
 

dvc/commands/data_sync.py

@@ -194,7 +194,7 @@ def add_parser(subparsers, _parent_parser):
     )
     pull_parser.add_argument(
         "--run-cache",
-        action="store_true",
+        action=argparse.BooleanOptionalAction,
         default=False,
         help="Fetch run history for all stages.",
     )
@@ -262,7 +262,7 @@ def add_parser(subparsers, _parent_parser):
     )
     push_parser.add_argument(
         "--run-cache",
-        action="store_true",
+        action=argparse.BooleanOptionalAction,
         default=False,
         help="Push run history for all stages.",
     )
@@ -324,7 +324,7 @@ def add_parser(subparsers, _parent_parser):
     )
     fetch_parser.add_argument(
         "--run-cache",
-        action="store_true",
+        action=argparse.BooleanOptionalAction,
         default=False,
         help="Fetch run history for all stages.",
     )

dvc/commands/dataset.py

@@ -34,7 +34,7 @@ class CmdDatasetAdd(CmdBase):
 
         url = dataset.spec.url
         ver: str = ""
-        if dataset.type == "dvcx":
+        if dataset.type == "dc":
             ver = f"v{dataset.lock.version}"
         if dataset.type == "dvc":
             if dataset.lock.path:
@@ -42,7 +42,7 @@ class CmdDatasetAdd(CmdBase):
             if rev := dataset.lock.rev:
                 ver = rev
 
-        ver_part: Optional["Text"] = None
+        ver_part: Optional[Text] = None
         if ver:
             ver_part = ui.rich_text.assemble(" @ ", (ver, "repr.number"))
         text = ui.rich_text.assemble("(", (url, "repr.url"), ver_part or "", ")")
@@ -55,7 +55,7 @@ class CmdDatasetAdd(CmdBase):
             raise DvcException("--path can't be used without --dvc")
 
         d = vars(self.args)
-        for key in ["dvc", "dvcx", "url"]:
+        for key in ["dvc", "dc", "url"]:
             if url := d.pop(key, None):
                 d.update({"type": key, "url": url})
                 break
@@ -88,8 +88,8 @@ class CmdDatasetUpdate(CmdBase):
         assert new.lock
 
         v: Optional[tuple[str, str]] = None
-        if dataset.type == "dvcx":
-            assert new.type == "dvcx"
+        if dataset.type == "dc":
+            assert new.type == "dc"
             if new.lock.version < dataset.lock.version:
                 action = "Downgrading"
 
@@ -168,7 +168,7 @@ def add_parser(subparsers, parent_parser):
 
     url_exclusive_group = ds_add_parser.add_mutually_exclusive_group(required=True)
     url_exclusive_group.add_argument(
-        "--dvcx", metavar="name", help="Name of the dvcx dataset to track"
+        "--dc", metavar="name", help="Name of the DataChain dataset to track"
     )
     url_exclusive_group.add_argument(
         "--dvc",
@@ -219,7 +219,7 @@ remote://remote_name/path/to/file/or/dir (see `dvc remote`)
         "--rev",
         "--version",
         nargs="?",
-        help="DVCX dataset version or Git revision (e.g. SHA, branch, tag)",
+        help="DataChain dataset version or Git revision (e.g. SHA, branch, tag)",
         metavar="<version>",
     )
     ds_update_parser.set_defaults(func=CmdDatasetUpdate)

dvc/commands/experiments/__init__.py

@@ -58,6 +58,15 @@ def add_parser(subparsers, parent_parser):
     hide_subparsers_from_help(experiments_subparsers)
 
 
+def add_keep_selection_flag(experiments_subcmd_parser):
+    experiments_subcmd_parser.add_argument(
+        "--keep",
+        action="store_true",
+        default=False,
+        help="Keep the selected experiments instead of removing them.",
+    )
+
+
 def add_rev_selection_flags(
     experiments_subcmd_parser, command: str, default: bool = True
 ):

dvc/commands/experiments/pull.py

@@ -1,3 +1,5 @@
+import argparse
+
 from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
@@ -81,7 +83,7 @@ def add_parser(experiments_subparsers, parent_parser):
     )
     experiments_pull_parser.add_argument(
         "--run-cache",
-        action="store_true",
+        action=argparse.BooleanOptionalAction,
         default=False,
         help="Pull run history for all stages.",
     )

dvc/commands/experiments/push.py

@@ -1,3 +1,4 @@
+import argparse
 from typing import Any
 
 from dvc.cli import completion, formatter
@@ -120,7 +121,7 @@ def add_parser(experiments_subparsers, parent_parser):
     )
     experiments_push_parser.add_argument(
         "--run-cache",
-        action="store_true",
+        action=argparse.BooleanOptionalAction,
         default=False,
         help="Push run history for all stages.",
     )

dvc/commands/experiments/remove.py

@@ -34,6 +34,7 @@ class CmdExperimentsRemove(CmdBase):
             num=self.args.num,
             queue=self.args.queue,
             git_remote=self.args.git_remote,
+            keep=self.args.keep,
         )
         if removed:
             ui.write(f"Removed experiments: {humanize.join(map(repr, removed))}")
@@ -44,7 +45,7 @@ class CmdExperimentsRemove(CmdBase):
 
 
 def add_parser(experiments_subparsers, parent_parser):
-    from . import add_rev_selection_flags
+    from . import add_keep_selection_flag, add_rev_selection_flags
 
     EXPERIMENTS_REMOVE_HELP = "Remove experiments."
     experiments_remove_parser = experiments_subparsers.add_parser(
@@ -57,6 +58,7 @@ def add_parser(experiments_subparsers, parent_parser):
     )
     remove_group = experiments_remove_parser.add_mutually_exclusive_group()
     add_rev_selection_flags(experiments_remove_parser, "Remove", False)
+    add_keep_selection_flag(experiments_remove_parser)
     remove_group.add_argument(
         "--queue", action="store_true", help="Remove all queued experiments."
     )

dvc/commands/gc.py

@@ -54,13 +54,13 @@ class CmdGC(CmdBase):
             msg += " and all experiments"
 
         if self.args.not_in_remote:
-            msg += " that are present in the DVC remote"
+            msg += " that are not present in the DVC remote"
 
         if self.args.repos:
             msg += " of the current and the following repos:"
 
             for repo_path in self.args.repos:
-                msg += "\n  - %s" % os.path.abspath(repo_path)
+                msg += f"\n  - {os.path.abspath(repo_path)}"
         else:
             msg += " of the current repo."
 

dvc/commands/ls/__init__.py

@@ -1,3 +1,5 @@
+from typing import Callable
+
 from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.utils import DictAction, append_doc_link
@@ -9,54 +11,178 @@ from dvc.ui import ui
 logger = logger.getChild(__name__)
 
 
-def _format_entry(entry, fmt):
-    from dvc.utils.humanize import naturalsize
+def _get_formatter(with_color: bool = False) -> Callable[[dict], str]:
+    def fmt(entry: dict) -> str:
+        return entry["path"]
 
-    size = entry.get("size")
-    if size is None:
-        size = ""
-    else:
-        size = naturalsize(size)
-    return size, fmt(entry)
-
-
-def show_entries(entries, with_color=False, with_size=False):
     if with_color:
         ls_colors = LsColors()
-        fmt = ls_colors.format
-    else:
+        return ls_colors.format
 
-        def fmt(entry):
-            return entry["path"]
+    return fmt
 
+
+def _format_entry(entry, name, with_size=True, with_hash=False):
+    from dvc.utils.humanize import naturalsize
+
+    ret = []
     if with_size:
-        ui.table([_format_entry(entry, fmt) for entry in entries])
+        size = entry.get("size")
+        if size is None or (size <= 0 and entry.get("isdir")):
+            size = ""
+        else:
+            size = naturalsize(size)
+        ret.append(size)
+    if with_hash:
+        md5 = entry.get("md5", "")
+        ret.append(md5)
+    ret.append(name)
+    return ret
+
+
+def show_entries(entries, with_color=False, with_size=False, with_hash=False):
+    fmt = _get_formatter(with_color)
+    if with_size or with_hash:
+        colalign = ("right",) if with_size else None
+        ui.table(
+            [
+                _format_entry(
+                    entry,
+                    fmt(entry),
+                    with_size=with_size,
+                    with_hash=with_hash,
+                )
+                for entry in entries
+            ],
+            colalign=colalign,
+        )
         return
 
     # NOTE: this is faster than ui.table for very large number of entries
     ui.write("\n".join(fmt(entry) for entry in entries))
 
 
+class TreePart:
+    Edge = "├── "
+    Line = "│   "
+    Corner = "└── "
+    Blank = "    "
+
+
+def _build_tree_structure(
+    entries, with_color=False, with_size=False, with_hash=False, _depth=0, _prefix=""
+):
+    rows = []
+    fmt = _get_formatter(with_color)
+
+    num_entries = len(entries)
+    for i, (name, entry) in enumerate(entries.items()):
+        entry["path"] = name
+        is_last = i >= num_entries - 1
+        tree_part = ""
+        if _depth > 0:
+            tree_part = TreePart.Corner if is_last else TreePart.Edge
+
+        row = _format_entry(
+            entry,
+            _prefix + tree_part + fmt(entry),
+            with_size=with_size,
+            with_hash=with_hash,
+        )
+        rows.append(row)
+
+        if contents := entry.get("contents"):
+            new_prefix = _prefix
+            if _depth > 0:
+                new_prefix += TreePart.Blank if is_last else TreePart.Line
+            new_rows = _build_tree_structure(
+                contents,
+                with_color=with_color,
+                with_size=with_size,
+                with_hash=with_hash,
+                _depth=_depth + 1,
+                _prefix=new_prefix,
+            )
+            rows.extend(new_rows)
+
+    return rows
+
+
+def show_tree(entries, with_color=False, with_size=False, with_hash=False):
+    import tabulate
+
+    rows = _build_tree_structure(
+        entries,
+        with_color=with_color,
+        with_size=with_size,
+        with_hash=with_hash,
+    )
+
+    colalign = ("right",) if with_size else None
+
+    _orig = tabulate.PRESERVE_WHITESPACE
+    tabulate.PRESERVE_WHITESPACE = True
+    try:
+        ui.table(rows, colalign=colalign)
+    finally:
+        tabulate.PRESERVE_WHITESPACE = _orig
+
+
 class CmdList(CmdBaseNoRepo):
-    def run(self):
+    def _show_tree(self):
+        from dvc.repo.ls import ls_tree
+
+        entries = ls_tree(
+            self.args.url,
+            self.args.path,
+            rev=self.args.rev,
+            dvc_only=self.args.dvc_only,
+            config=self.args.config,
+            remote=self.args.remote,
+            remote_config=self.args.remote_config,
+            maxdepth=self.args.level,
+        )
+        show_tree(
+            entries,
+            with_color=True,
+            with_size=self.args.size,
+            with_hash=self.args.show_hash,
+        )
+        return 0
+
+    def _show_list(self):
         from dvc.repo import Repo
 
-        try:
-            entries = Repo.ls(
-                self.args.url,
-                self.args.path,
-                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,
+        entries = Repo.ls(
+            self.args.url,
+            self.args.path,
+            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,
+            maxdepth=self.args.level,
+        )
+        if self.args.json:
+            ui.write_json(entries)
+        elif entries:
+            show_entries(
+                entries,
+                with_color=True,
+                with_size=self.args.size,
+                with_hash=self.args.show_hash,
             )
-            if self.args.json:
-                ui.write_json(entries)
-            elif entries:
-                show_entries(entries, with_color=True, with_size=self.args.size)
-            return 0
+        return 0
+
+    def run(self):
+        if self.args.tree and self.args.json:
+            raise DvcException("Cannot use --tree and --json options together.")
+
+        try:
+            if self.args.tree:
+                return self._show_tree()
+            return self._show_list()
         except FileNotFoundError:
             logger.exception("")
             return 1
@@ -85,6 +211,19 @@ def add_parser(subparsers, parent_parser):
         action="store_true",
         help="Recursively list files.",
     )
+    list_parser.add_argument(
+        "-T",
+        "--tree",
+        action="store_true",
+        help="Recurse into directories as a tree.",
+    )
+    list_parser.add_argument(
+        "-L",
+        "--level",
+        metavar="depth",
+        type=int,
+        help="Limit the depth of recursion.",
+    )
     list_parser.add_argument(
         "--dvc-only", action="store_true", help="Show only DVC outputs."
     )
@@ -123,6 +262,12 @@ def add_parser(subparsers, parent_parser):
         ),
     )
     list_parser.add_argument("--size", action="store_true", help="Show sizes.")
+    list_parser.add_argument(
+        "--show-hash",
+        help="Display hash value for each item.",
+        action="store_true",
+        default=False,
+    )
     list_parser.add_argument(
         "path",
         nargs="?",

dvc/commands/ls/ls_colors.py

@@ -32,7 +32,9 @@ class LsColors:
         if entry.get("isexec", False):
             return self._format(text, code="ex")
 
-        _, ext = os.path.splitext(text)
+        stem, ext = os.path.splitext(text)
+        if not ext and stem.startswith("."):
+            ext = stem
         return self._format(text, ext=ext)
 
     def _format(self, text, code=None, ext=None):

dvc/commands/ls_url.py

@@ -3,26 +3,45 @@ from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.utils import DictAction, append_doc_link
 from dvc.log import logger
 
-from .ls import show_entries
+from .ls import show_entries, show_tree
 
 logger = logger.getChild(__name__)
 
 
 class CmdListUrl(CmdBaseNoRepo):
-    def run(self):
-        from dvc.config import Config
+    def _show_tree(self, config):
+        from dvc.fs import parse_external_url
+        from dvc.repo.ls import _ls_tree
+
+        fs, fs_path = parse_external_url(
+            self.args.url, fs_config=self.args.fs_config, config=config
+        )
+        entries = _ls_tree(fs, fs_path, maxdepth=self.args.level)
+        show_tree(entries, with_color=True, with_size=self.args.size)
+        return 0
+
+    def _show_list(self, config):
         from dvc.repo import Repo
 
         entries = Repo.ls_url(
             self.args.url,
             recursive=self.args.recursive,
+            maxdepth=self.args.level,
             fs_config=self.args.fs_config,
-            config=Config.from_cwd(),
+            config=config,
         )
         if entries:
             show_entries(entries, with_color=True, with_size=self.args.size)
         return 0
 
+    def run(self):
+        from dvc.config import Config
+
+        config = Config.from_cwd()
+        if self.args.tree:
+            return self._show_tree(config=config)
+        return self._show_list(config=config)
+
 
 def add_parser(subparsers, parent_parser):
     LS_HELP = "List directory contents from URL."
@@ -40,6 +59,19 @@ def add_parser(subparsers, parent_parser):
     lsurl_parser.add_argument(
         "-R", "--recursive", action="store_true", help="Recursively list files."
     )
+    lsurl_parser.add_argument(
+        "-T",
+        "--tree",
+        action="store_true",
+        help="Recurse into directories as a tree.",
+    )
+    lsurl_parser.add_argument(
+        "-L",
+        "--level",
+        metavar="depth",
+        type=int,
+        help="Limit the depth of recursion.",
+    )
     lsurl_parser.add_argument("--size", action="store_true", help="Show sizes.")
     lsurl_parser.add_argument(
         "--fs-config",

dvc/commands/plots.py

@@ -72,7 +72,7 @@ class CmdPlots(CmdBase):
                 html_template_path = os.path.join(self.repo.dvc_dir, html_template_path)
         return html_template_path
 
-    def run(self) -> int:  # noqa: C901, PLR0911, PLR0912
+    def run(self) -> int:  # noqa: C901, PLR0911
         from pathlib import Path
 
         from dvc.render.match import match_defs_renderers

dvc/commands/remote.py

@@ -110,8 +110,16 @@ class CmdRemoteDefault(CmdRemote):
 class CmdRemoteList(CmdRemote):
     def run(self):
         conf = self.config.read(self.args.level)
+        default_remote = conf["core"].get("remote")
+
         for name, remote_conf in conf["remote"].items():
-            ui.write(name, remote_conf["url"], sep="\t")
+            if name == default_remote:
+                text = f"{name}\t{remote_conf['url']}\t(default)"
+                color = "green"
+            else:
+                text = f"{name}\t{remote_conf['url']}"
+                color = ""
+            ui.write(ui.rich_text(text, style=color), styled=True)
         return 0
 
 

dvc/commands/repro.py

@@ -117,10 +117,7 @@ and then the stage name name.
         "--pull",
         action="store_true",
         default=False,
-        help=(
-            "Try automatically pulling missing cache for outputs restored "
-            "from the run-cache."
-        ),
+        help="Try automatically pulling missing data.",
     )
     repro_parser.add_argument(
         "--allow-missing",

dvc/commands/stage.py

@@ -68,7 +68,7 @@ def prepare_stages_data(
 class CmdStageList(CmdBase):
     def _get_stages(self) -> Iterable["Stage"]:
         if self.args.all:
-            stages: list["Stage"] = self.repo.index.stages
+            stages: list[Stage] = self.repo.index.stages
             logger.trace("%d no. of stages found", len(stages))
             return stages
 

dvc/commands/studio.py

@@ -18,11 +18,16 @@ class CmdStudioLogin(CmdConfig):
         from dvc.utils.studio import STUDIO_URL
         from dvc_studio_client.auth import StudioAuthError, get_access_token
 
+        studio = self.config.get("studio", {})
         name = self.args.name
-        hostname = self.args.hostname or os.environ.get(DVC_STUDIO_URL) or STUDIO_URL
+        hostname = (
+            self.args.hostname
+            or os.environ.get(DVC_STUDIO_URL)
+            or studio.get("url")
+            or STUDIO_URL
+        )
         scopes = self.args.scopes
 
-        studio = self.config.get("studio", {})
         if studio.get("url", hostname) == hostname and "token" in studio:
             raise DvcException(
                 "Token already exists. "

dvc/compare.py

@@ -160,7 +160,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
             if key not in keys:
                 self.add_column(key)
 
-        row: list["CellT"] = [
+        row: list[CellT] = [
             with_value(d.get(key), self._fill_value) for key in self.keys()
         ]
         self.append(row)

dvc/config.py

@@ -1,6 +1,8 @@
 """DVC config objects."""
 
+import ntpath
 import os
+import posixpath
 import re
 from contextlib import contextmanager
 from functools import partial
@@ -268,12 +270,17 @@ class Config(dict):
         if re.match(r"\w+://", path):
             return path
 
+        if os.name == "nt" and posixpath.isabs(path) and ntpath.sep not in path:
+            return path
+
         if os.path.isabs(path):
             return path
 
         # on windows convert slashes to backslashes
         # to have path compatible with abs_conf_dir
         if os.path.sep == "\\" and "/" in path:
+            if path.startswith("/"):
+                path = path.replace("/", "\\\\", 1)
             path = path.replace("/", "\\")
 
         expanded = os.path.expanduser(path)
@@ -305,6 +312,9 @@ class Config(dict):
         if os.path.expanduser(path) != path:
             return localfs.as_posix(path)
 
+        if os.name == "nt" and posixpath.isabs(path) and ntpath.sep not in path:
+            return path
+
         if isinstance(path, RelPath) or not os.path.isabs(path):
             path = relpath(path, conf_dir)
             return localfs.as_posix(path)
@@ -330,6 +340,7 @@ class Config(dict):
                     "gdrive_user_credentials_file": func,
                     "gdrive_service_account_json_file_path": func,
                     "credentialpath": func,
+                    "configpath": func,
                     "keyfile": func,
                     "cert_path": func,
                     "key_path": func,

dvc/config_schema.py

@@ -193,6 +193,7 @@ SCHEMA = {
                     "session_token": str,
                     Optional("listobjects", default=False): Bool,  # obsoleted
                     Optional("use_ssl", default=True): Bool,
+                    Optional("allow_anonymous_login", default=False): Bool,
                     "ssl_verify": Any(Bool, str),
                     "sse": str,
                     "sse_kms_key_id": str,
@@ -214,6 +215,7 @@ SCHEMA = {
                     "credentialpath": str,
                     "endpointurl": str,
                     Optional("verify", default=False): Bool,
+                    Optional("allow_anonymous_login", default=False): Bool,
                     **REMOTE_COMMON,
                 },
                 "ssh": {
@@ -232,7 +234,12 @@ SCHEMA = {
                     Optional("verify", default=False): Bool,
                     **REMOTE_COMMON,
                 },
-                "hdfs": {"user": str, "kerb_ticket": str, **REMOTE_COMMON},
+                "hdfs": {
+                    "user": str,
+                    "kerb_ticket": str,
+                    "replication": int,
+                    **REMOTE_COMMON,
+                },
                 "webhdfs": {
                     "kerberos": Bool,
                     "kerberos_principal": str,

dvc/daemon.py

@@ -44,10 +44,10 @@ def _win_detached_subprocess(args: Sequence[str], **kwargs) -> int:
 
     startupinfo = STARTUPINFO()
     startupinfo.dwFlags |= STARTF_USESHOWWINDOW
-    popen = subprocess.Popen(
+    popen = subprocess.Popen(  # noqa: S603
         args,
         close_fds=True,
-        shell=False,  # noqa: S603
+        shell=False,
         startupinfo=startupinfo,
         creationflags=creationflags,
         **kwargs,
@@ -149,7 +149,7 @@ def _spawn(
     env: Optional[Mapping[str, str]] = None,
     output_file: Optional[str] = None,
 ) -> int:
-    file: "AbstractContextManager[Any]" = nullcontext()
+    file: AbstractContextManager[Any] = nullcontext()
     kwargs = {}
     if output_file:
         file = open(output_file, "ab")  # noqa: SIM115
@@ -164,7 +164,7 @@ def _spawn(
         return _detached_subprocess(executable + args, env=env, **kwargs)
 
 
-def daemonize(args: list[str], executable: Union[None, str, list[str]] = None) -> None:
+def daemonize(args: list[str], executable: Union[str, list[str], None] = None) -> None:
     if os.name not in ("posix", "nt"):
         return
 

dvc/dagascii.py

@@ -104,21 +104,21 @@ class AsciiCanvas:
                 if dx == 0:
                     y = y0
                 else:
-                    y = y0 + int(round((x - x0) * dy / float(dx)))
+                    y = y0 + round((x - x0) * dy / float(dx))
                 self.point(x, y, char)
         elif y0 < y1:
             for y in range(y0, y1 + 1):
                 if dy == 0:
                     x = x0
                 else:
-                    x = x0 + int(round((y - y0) * dx / float(dy)))
+                    x = x0 + round((y - y0) * dx / float(dy))
                 self.point(x, y, char)
         else:
             for y in range(y1, y0 + 1):
                 if dy == 0:
                     x = x0
                 else:
-                    x = x1 + int(round((y - y1) * dx / float(dy)))
+                    x = x1 + round((y - y1) * dx / float(dy))
                 self.point(x, y, char)
 
     def text(self, x, y, text):
@@ -258,8 +258,8 @@ def draw(vertices, edges):
     maxx = max(Xs)
     maxy = max(Ys)
 
-    canvas_cols = int(math.ceil(math.ceil(maxx) - math.floor(minx))) + 1
-    canvas_lines = int(round(maxy - miny))
+    canvas_cols = math.ceil(math.ceil(maxx) - math.floor(minx)) + 1
+    canvas_lines = round(maxy - miny)
 
     canvas = AsciiCanvas(canvas_cols, canvas_lines)
 
@@ -270,10 +270,10 @@ def draw(vertices, edges):
             start = edge.view._pts[index - 1]
             end = edge.view._pts[index]
 
-            start_x = int(round(start[0] - minx))
-            start_y = int(round(start[1] - miny))
-            end_x = int(round(end[0] - minx))
-            end_y = int(round(end[1] - miny))
+            start_x = round(start[0] - minx)
+            start_y = round(start[1] - miny)
+            end_x = round(end[0] - minx)
+            end_y = round(end[1] - miny)
 
             assert start_x >= 0
             assert start_y >= 0
@@ -288,12 +288,12 @@ def draw(vertices, edges):
         y = vertex.view.xy[1]
 
         canvas.box(
-            int(round(x - minx)),
-            int(round(y - miny)),
+            round(x - minx),
+            round(y - miny),
             vertex.view.w,
             vertex.view.h,
         )
 
-        canvas.text(int(round(x - minx)) + 1, int(round(y - miny)) + 1, vertex.data)
+        canvas.text(round(x - minx) + 1, round(y - miny) + 1, vertex.data)
 
     return canvas.draw()

dvc/dependency/base.py

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

dvc/dependency/db.py

@@ -37,13 +37,17 @@ class AbstractDependency(Dependency):
     """Dependency without workspace/fs/fs_path"""
 
     def __init__(self, stage: "Stage", info: dict[str, Any], *args, **kwargs):
-        self.repo: "Repo" = stage.repo
+        self.repo: Repo = stage.repo
         self.stage = stage
         self.fs = None
         self.fs_path = None
         self.def_path = None  # type: ignore[assignment]
         self.info = info or {}
 
+    @property
+    def is_in_repo(self):
+        return False
+
 
 class DbDependency(AbstractDependency):
     PARAM_CONNECTION = "connection"
@@ -70,8 +74,8 @@ class DbDependency(AbstractDependency):
         return self.db_info.get(self.PARAM_QUERY) or self.db_info.get(self.PARAM_TABLE)
 
     def __repr__(self):
-        return "{}:{}".format(
-            self.__class__.__name__, "".join(f"{k}=={v}" for k, v in self.info.items())
+        return "{}: {}".format(
+            self.__class__.__name__, "".join(f"{k}={v}" for k, v in self.info.items())
         )
 
     def __str__(self):

dvc/dependency/repo.py

@@ -9,7 +9,9 @@ from .base import Dependency
 
 if TYPE_CHECKING:
     from dvc.fs import DVCFileSystem
+    from dvc.output import Output
     from dvc.stage import Stage
+    from dvc_data.hashfile.hash_info import HashInfo
 
 
 class RepoDependency(Dependency):
@@ -35,7 +37,7 @@ class RepoDependency(Dependency):
         super().__init__(stage, *args, **kwargs)
 
         self.fs = self._make_fs()
-        self.fs_path = as_posix(self.def_path)
+        self.fs_path = as_posix(self.fs.normpath(self.def_path))
 
     def _parse_path(self, fs, fs_path):  # noqa: ARG002
         return None
@@ -91,6 +93,27 @@ class RepoDependency(Dependency):
             self.PARAM_REPO: self._dump_def_repo(self.def_repo),
         }
 
+    def download(self, to: "Output", jobs: Optional[int] = None):
+        from dvc.fs import LocalFileSystem
+
+        files = super().download(to=to, jobs=jobs)
+        if not isinstance(to.fs, LocalFileSystem):
+            return
+
+        hashes: list[tuple[str, HashInfo, dict[str, Any]]] = []
+        for src_path, dest_path, maybe_info in files:
+            try:
+                info = maybe_info or self.fs.info(src_path)
+                hash_info = info["dvc_info"]["entry"].hash_info
+                dest_info = to.fs.info(dest_path)
+            except (KeyError, AttributeError):
+                # If no hash info found, just keep going and output will be hashed later
+                continue
+            if hash_info:
+                hashes.append((dest_path, hash_info, dest_info))
+        cache = to.cache if to.use_cache else to.local_cache
+        cache.state.save_many(hashes, to.fs)
+
     def update(self, rev: Optional[str] = None):
         if rev:
             self.def_repo[self.PARAM_REV] = rev
@@ -123,11 +146,15 @@ class RepoDependency(Dependency):
         else:
             config = Config.load_file(conf)
 
+        # Setup config to the new DVCFileSystem to use the remote repo, but rely on the
+        # local cache instead of the remote's cache. This avoids re-streaming of data,
+        # but messes up the call to `_get_remote_config()` downstream, which will need
+        # to ignore cache parameters.
         config["cache"] = self.repo.config["cache"]
         config["cache"]["dir"] = self.repo.cache.local_cache_dir
 
         return DVCFileSystem(
-            url=self.def_repo[self.PARAM_URL],
+            repo=self.def_repo[self.PARAM_URL],
             rev=rev or self._get_rev(locked=locked),
             subrepos=True,
             config=config,

dvc/dirs.py

@@ -1,4 +1,5 @@
 import os
+from typing import Optional
 
 import platformdirs
 
@@ -20,7 +21,24 @@ def global_config_dir():
     )
 
 
-def site_cache_dir():
-    return os.getenv(env.DVC_SITE_CACHE_DIR) or platformdirs.site_cache_dir(
-        APPNAME, APPAUTHOR, opinion=True
-    )
+def site_cache_dir(config_site_cache_dir: Optional[str] = None):
+    from platformdirs import PlatformDirs
+    from platformdirs.unix import Unix
+
+    if dvc_site_cache_dir := os.getenv(env.DVC_SITE_CACHE_DIR):
+        return dvc_site_cache_dir
+
+    if config_site_cache_dir is not None:
+        return config_site_cache_dir
+
+    if issubclass(Unix, PlatformDirs):
+        # Return the cache directory shared by users, e.g. `/var/tmp/$appname`
+        # NOTE: platformdirs>=5 changed `site_cache_dir` to return /var/cache/$appname.
+        # as the following path is considered insecure.
+        # For details, see: https://github.com/tox-dev/platformdirs/pull/239
+
+        # FIXME: keeping the old behavior temporarily to avoid dependency conflict.
+        #        In the future, consider migrating to a more secure directory.
+        return f"/var/tmp/{APPNAME}"  # noqa: S108
+
+    return platformdirs.site_cache_dir(APPNAME, APPAUTHOR, opinion=True)

dvc/fs/__init__.py

@@ -1,3 +1,5 @@
+import glob
+from itertools import repeat
 from typing import Optional
 from urllib.parse import urlparse
 
@@ -18,7 +20,7 @@ from dvc_objects.fs import (  # noqa: F401
     system,
     utils,
 )
-from dvc_objects.fs.base import AnyFSPath, FileSystem  # noqa: F401, TCH001
+from dvc_objects.fs.base import AnyFSPath, FileSystem  # noqa: F401, TC001
 from dvc_objects.fs.errors import (  # noqa: F401
     AuthError,
     ConfigError,
@@ -46,12 +48,24 @@ known_implementations.update(
 
 def download(
     fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None
-) -> int:
+) -> list[tuple[str, str, Optional[dict]]]:
     from dvc.scm import lfs_prefetch
 
     from .callbacks import TqdmCallback
 
     with TqdmCallback(desc=f"Downloading {fs.name(fs_path)}", unit="files") as cb:
+        if isinstance(fs, DVCFileSystem):
+            lfs_prefetch(
+                fs,
+                [
+                    f"{fs.normpath(glob.escape(fs_path))}/**"
+                    if fs.isdir(fs_path)
+                    else glob.escape(fs_path)
+                ],
+            )
+            if not glob.has_magic(fs_path):
+                return fs._get(fs_path, to, 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):
@@ -60,7 +74,7 @@ def download(
             ]
             if not from_infos:
                 localfs.makedirs(to, exist_ok=True)
-                return 0
+                return []
             to_infos = [
                 localfs.join(to, *fs.relparts(info, fs_path)) for info in from_infos
             ]
@@ -68,12 +82,10 @@ def download(
             from_infos = [fs_path]
             to_infos = [to]
 
-        if isinstance(fs, DVCFileSystem):
-            lfs_prefetch(fs, from_infos)
         cb.set_size(len(from_infos))
         jobs = jobs or fs.jobs
         generic.copy(fs, from_infos, localfs, to_infos, callback=cb, batch_size=jobs)
-        return len(to_infos)
+        return list(zip(from_infos, to_infos, repeat(None)))
 
 
 def parse_external_url(url, fs_config=None, config=None):

dvc/fs/dvc.py

@@ -4,22 +4,28 @@ import ntpath
 import os
 import posixpath
 import threading
-from collections import deque
-from contextlib import ExitStack, suppress
+from collections import defaultdict, deque
+from contextlib import ExitStack, nullcontext, suppress
+from glob import has_magic
 from typing import TYPE_CHECKING, Any, Callable, Optional, Union
 
-from fsspec.spec import AbstractFileSystem
+from fsspec.spec import DEFAULT_CALLBACK, AbstractFileSystem
 from funcy import wrap_with
 
 from dvc.log import logger
-from dvc_objects.fs.base import FileSystem
+from dvc.utils.threadpool import ThreadPoolExecutor
+from dvc_objects.fs.base import AnyFSPath, FileSystem
 
 from .data import DataFileSystem
 
 if TYPE_CHECKING:
+    from contextlib import AbstractContextManager
+
     from dvc.repo import Repo
     from dvc.types import DictStrAny, StrPath
 
+    from .callbacks import Callback
+
 logger = logger.getChild(__name__)
 
 RepoFactory = Union[Callable[..., "Repo"], type["Repo"]]
@@ -77,11 +83,10 @@ class _DVCFileSystem(AbstractFileSystem):
     cachable = False
     root_marker = "/"
 
-    def __init__(  # noqa: PLR0913
+    def __init__(
         self,
-        url: Optional[str] = None,
+        repo: Union["Repo", os.PathLike[str], str, None] = None,
         rev: Optional[str] = None,
-        repo: Optional["Repo"] = None,
         subrepos: bool = False,
         repo_factory: Optional[RepoFactory] = None,
         fo: Optional[str] = None,
@@ -95,7 +100,8 @@ class _DVCFileSystem(AbstractFileSystem):
         """DVC + git-tracked files fs.
 
         Args:
-            path (str, optional): URL or path to a DVC/Git repository.
+            repo (str | os.PathLike[str] | Repo, optional): A url or a path to a DVC/Git
+                repository, or a `Repo` instance.
                 Defaults to a DVC repository in the current working directory.
                 Both HTTP and SSH protocols are supported for remote Git repos
                 (e.g. [user@]server:project.git).
@@ -105,7 +111,6 @@ class _DVCFileSystem(AbstractFileSystem):
                 In case of a local repository, if rev is unspecified, it will
                 default to the working directory.
                 If the repo is not a Git repo, this option is ignored.
-            repo (:obj:`Repo`, optional): `Repo` instance.
             subrepos (bool): traverse to subrepos.
                 By default, it ignores subrepos.
             repo_factory (callable): A function to initialize subrepo with.
@@ -130,13 +135,23 @@ class _DVCFileSystem(AbstractFileSystem):
             ...    rev="main",
             ... )
         """
+        from dvc.repo import Repo
+
+        # kwargs.get("url") is for maintaining backward compatibility
+        repo = repo or fo or kwargs.get("url")
+        if isinstance(repo, Repo):
+            self._repo: Optional[Repo] = repo
+            url = None
+        else:
+            self._repo = None
+            url = os.fspath(repo) if repo else None
+
         super().__init__()
-        self._repo = repo
         self._repo_factory = repo_factory
         self._traverse_subrepos = subrepos
         self._repo_stack = ExitStack()
         self._repo_kwargs = {
-            "url": url if url is not None else fo,
+            "url": url,
             "rev": rev,
             "subrepos": subrepos,
             "config": config,
@@ -360,7 +375,7 @@ class _DVCFileSystem(AbstractFileSystem):
         except FileNotFoundError:
             return False
 
-    def ls(self, path, detail=True, dvc_only=False, **kwargs):  # noqa: C901
+    def ls(self, path, detail=True, dvc_only=False, **kwargs):  # noqa: C901, PLR0912
         key = self._get_key_from_relative(path)
         repo, dvc_fs, subkey = self._get_subrepo_info(key)
 
@@ -429,7 +444,7 @@ class _DVCFileSystem(AbstractFileSystem):
         ignore_subrepos = kwargs.get("ignore_subrepos", True)
         return self._info(key, path, ignore_subrepos=ignore_subrepos)
 
-    def _info(  # noqa: C901, PLR0912
+    def _info(  # noqa: C901
         self, key, path, ignore_subrepos=True, check_ignored=True
     ):
         repo, dvc_fs, subkey = self._get_subrepo_info(key)
@@ -474,9 +489,128 @@ class _DVCFileSystem(AbstractFileSystem):
         info["name"] = path
         return info
 
+    def get(
+        self,
+        rpath,
+        lpath,
+        recursive=False,
+        callback=DEFAULT_CALLBACK,
+        maxdepth=None,
+        batch_size=None,
+        **kwargs,
+    ):
+        self._get(
+            rpath,
+            lpath,
+            recursive=recursive,
+            callback=callback,
+            maxdepth=maxdepth,
+            batch_size=batch_size,
+            **kwargs,
+        )
+
+    def _get(  # noqa: C901, PLR0912, PLR0915
+        self,
+        rpath,
+        lpath,
+        recursive=False,
+        callback=DEFAULT_CALLBACK,
+        maxdepth=None,
+        batch_size=None,
+        **kwargs,
+    ) -> list[tuple[str, str, Optional[dict]]]:
+        if (
+            isinstance(rpath, list)
+            or isinstance(lpath, list)
+            or has_magic(rpath)
+            or not self.exists(rpath)
+            or not recursive
+        ):
+            super().get(
+                rpath,
+                lpath,
+                recursive=recursive,
+                callback=callback,
+                maxdepth=maxdepth,
+                **kwargs,
+            )
+            return []
+
+        if os.path.isdir(lpath) or lpath.endswith(os.path.sep):
+            lpath = self.join(lpath, os.path.basename(rpath))
+
+        if self.isfile(rpath):
+            with callback.branched(rpath, lpath) as child:
+                self.get_file(rpath, lpath, callback=child, **kwargs)
+                return [(rpath, lpath, None)]
+
+        result: list[tuple[str, str, Optional[dict]]] = []
+        _dirs: list[str] = []
+        _files: dict[FileSystem, list[tuple[str, str, Optional[dict]]]]
+        _files = defaultdict(list)
+
+        for root, dirs, files in self.walk(rpath, maxdepth=maxdepth, detail=True):
+            if files:
+                callback.set_size((callback.size or 0) + len(files))
+
+            parts = self.relparts(root, rpath)
+            if parts in ((os.curdir,), ("",)):
+                parts = ()
+            dest_root = os.path.join(lpath, *parts)
+            if not maxdepth or len(parts) < maxdepth - 1:
+                _dirs.extend(f"{dest_root}{os.path.sep}{d}" for d in dirs)
+
+            key = self._get_key_from_relative(root)
+            _, dvc_fs, _ = self._get_subrepo_info(key)
+
+            for name, info in files.items():
+                dvc_info = info.get("dvc_info")
+                fs_info = info.get("fs_info")
+                if dvc_fs and dvc_info and not fs_info:
+                    fs = dvc_fs
+                    fs_path = dvc_info["name"]
+                else:
+                    fs = self.repo.fs
+                    fs_path = fs_info["name"]
+
+                src_path = f"{root}{self.sep}{name}"
+                dest_path = f"{dest_root}{os.path.sep}{name}"
+                _files[fs].append((fs_path, dest_path, dvc_info))
+                result.append((src_path, dest_path, info))
+
+        os.makedirs(lpath, exist_ok=True)
+        for d in _dirs:
+            os.makedirs(d, exist_ok=True)
+
+        def get_file(arg: tuple[FileSystem, tuple[str, str, Optional[dict]]]):
+            fs, (src, dest, info) = arg
+            kw = kwargs
+            if isinstance(fs, DataFileSystem):
+                kw = kw | {"info": info}
+            with callback.branched(src, dest) as child:
+                fs.get_file(src, dest, callback=child, **kw)
+
+        if batch_size == 1:
+            ctx: AbstractContextManager = nullcontext()
+            map_fn: Callable = map
+        else:
+            ctx = ThreadPoolExecutor(max_workers=batch_size)
+            map_fn = ctx.imap_unordered
+
+        with ctx:
+            it = ((fs, f) for fs, files in _files.items() for f in files)
+            deque(callback.wrap(map_fn(get_file, it)), maxlen=0)
+        return result
+
     def get_file(self, rpath, lpath, **kwargs):
+        dvc_info = kwargs.pop("info", {}).pop("dvc_info", None)
         key = self._get_key_from_relative(rpath)
         fs_path = self._from_key(key)
+        dirpath = os.path.dirname(lpath)
+        if dirpath:
+            # makedirs raises error if the string is empty
+            os.makedirs(dirpath, exist_ok=True)
+
         try:
             return self.repo.fs.get_file(fs_path, lpath, **kwargs)
         except FileNotFoundError:
@@ -485,7 +619,7 @@ class _DVCFileSystem(AbstractFileSystem):
                 raise
 
         dvc_path = _get_dvc_path(dvc_fs, subkey)
-        return dvc_fs.get_file(dvc_path, lpath, **kwargs)
+        return dvc_fs.get_file(dvc_path, lpath, info=dvc_info, **kwargs)
 
     def du(self, path, total=True, maxdepth=None, withdirs=False, **kwargs):
         if maxdepth is not None:
@@ -553,6 +687,45 @@ class DVCFileSystem(FileSystem):
     def getcwd(self):
         return self.fs.getcwd()
 
+    def _get(
+        self,
+        from_info: Union[AnyFSPath, list[AnyFSPath]],
+        to_info: Union[AnyFSPath, list[AnyFSPath]],
+        callback: "Callback" = DEFAULT_CALLBACK,
+        recursive: bool = False,
+        batch_size: Optional[int] = None,
+        **kwargs,
+    ) -> list[tuple[str, str, Optional[dict]]]:
+        # FileSystem.get is non-recursive by default if arguments are lists
+        # otherwise, it's recursive.
+        recursive = not (isinstance(from_info, list) and isinstance(to_info, list))
+        return self.fs._get(
+            from_info,
+            to_info,
+            callback=callback,
+            recursive=recursive,
+            batch_size=batch_size,
+            **kwargs,
+        )
+
+    def get(
+        self,
+        from_info: Union[AnyFSPath, list[AnyFSPath]],
+        to_info: Union[AnyFSPath, list[AnyFSPath]],
+        callback: "Callback" = DEFAULT_CALLBACK,
+        recursive: bool = False,
+        batch_size: Optional[int] = None,
+        **kwargs,
+    ) -> None:
+        self._get(
+            from_info,
+            to_info,
+            callback=callback,
+            batch_size=batch_size,
+            recursive=recursive,
+            **kwargs,
+        )
+
     @property
     def fsid(self) -> str:
         return self.fs.fsid
@@ -569,9 +742,10 @@ class DVCFileSystem(FileSystem):
         return self.fs.repo_url
 
     def from_os_path(self, path: str) -> str:
-        if os.path.isabs(path):
+        if os.path.isabs(path) or (
+            os.name == "nt" and posixpath.isabs(path) and ntpath.sep not in path
+        ):
             path = os.path.relpath(path, self.repo.root_dir)
-
         return as_posix(path)
 
     def close(self):

dvc/info.py

@@ -123,7 +123,7 @@ def _get_supported_remotes():
         if not fs_cls.get_missing_deps():
             dependencies = []
             for requirement in fs_cls.REQUIRES:
-                dependencies.append(
+                dependencies.append(  # noqa: PERF401
                     f"{requirement} = {importlib_metadata.version(requirement)}"
                 )
 

dvc/lock.py

@@ -4,6 +4,7 @@ import hashlib
 import os
 from abc import ABC, abstractmethod
 from datetime import timedelta
+from typing import Optional, Union
 
 import flufl.lock
 import zc.lockfile
@@ -174,9 +175,9 @@ class HardlinkLock(flufl.lock.Lock, LockBase):
         self._owned = True
         self._retry_errnos = []
 
-    def lock(self):
+    def lock(self, timeout: Optional[Union[timedelta, int]] = None):
         try:
-            super().lock(timedelta(seconds=DEFAULT_TIMEOUT))
+            super().lock(timeout or timedelta(seconds=DEFAULT_TIMEOUT))
         except flufl.lock.TimeOutError:
             raise LockError(FAILED_TO_LOCK_MESSAGE)  # noqa: B904
 

dvc/logger.py

@@ -36,7 +36,7 @@ def add_logging_level(level_name, level_num, method_name=None):
             self._log(level_num, message, args, **kwargs)
 
     def log_to_root(message, *args, **kwargs):
-        logging.log(level_num, message, *args, **kwargs)
+        logging.log(level_num, message, *args, **kwargs)  # noqa: LOG015
 
     # getLevelName resolves the numeric log level if already defined,
     # otherwise returns a string

dvc/output.py

@@ -390,7 +390,7 @@ class Output:
         self.can_push = push
 
         self.fs_path = self._parse_path(self.fs, fs_path)
-        self.obj: Optional["HashFile"] = None
+        self.obj: Optional[HashFile] = None
 
         self.remote = remote
 
@@ -431,7 +431,7 @@ class Output:
             self.meta.nfiles = len(self.files)
             self.meta.size = sum(filter(None, (f.get("size") for f in self.files)))
             self.meta.remote = first(f.get("remote") for f in self.files)
-        elif self.meta.nfiles or self.hash_info and self.hash_info.isdir:
+        elif self.meta.nfiles or (self.hash_info and self.hash_info.isdir):
             self.meta.isdir = True
             if not self.hash_info and self.hash_name not in ("md5", "md5-dos2unix"):
                 md5 = getattr(self.meta, "md5", None)
@@ -542,7 +542,9 @@ class Output:
             desc=f"Collecting files and computing hashes in {self}",
             disable=no_progress_bar,
         ) as pb:
-            return build(*args, callback=pb.as_callback(), **kwargs)
+            kwargs["callback"] = pb.as_callback()
+            kwargs.setdefault("checksum_jobs", self.fs.hash_jobs)
+            return build(*args, **kwargs)
 
     def _get_hash_meta(self):
         if self.use_cache:
@@ -730,6 +732,7 @@ class Output:
         from dvc_data.hashfile.checkout import LinkError, PromptError
 
         kwargs.setdefault("ignore", self.dvcignore)
+        kwargs.setdefault("checksum_jobs", self.fs.hash_jobs)
         try:
             return checkout(*args, **kwargs)
         except PromptError as exc:
@@ -749,11 +752,9 @@ class Output:
             granular = (
                 self.is_dir_checksum and filter_info and filter_info != self.fs_path
             )
-            # NOTE: trying to use hardlink during transfer only if we will be
-            # relinking later
-            hardlink = relink
+            hardlink = relink and next(iter(self.cache.cache_types), None) == "hardlink"
             if granular:
-                obj = self._commit_granular_dir(filter_info, hardlink)
+                obj = self._commit_granular_dir(filter_info, hardlink=hardlink)
             else:
                 staging, _, obj = self._build(
                     self.cache,
@@ -786,6 +787,7 @@ class Output:
                         state=self.repo.state,
                         prompt=prompt.confirm,
                         progress_callback=cb,
+                        old=obj,
                     )
                 self.set_exec()
 
@@ -902,7 +904,7 @@ class Output:
     def get_obj(
         self, filter_info: Optional[str] = None, **kwargs
     ) -> Optional["HashFile"]:
-        obj: Optional["HashFile"] = None
+        obj: Optional[HashFile] = None
         if self.obj:
             obj = self.obj
         elif self.files:
@@ -1158,7 +1160,7 @@ class Output:
             logger.warning(msg)
             return {}
 
-        obj: Optional["HashFile"]
+        obj: Optional[HashFile]
         if self.is_dir_checksum:
             obj = self._collect_used_dir_cache(**kwargs)
         else:
@@ -1353,7 +1355,7 @@ class Output:
         cache = self.cache if self.use_cache else self.local_cache
         assert isinstance(cache, HashFileDB)
 
-        new: "HashFile"
+        new: HashFile
         try:
             assert self.hash_name
             staging, meta, obj = self._build(
@@ -1365,7 +1367,7 @@ class Output:
                 dry_run=not self.use_cache,
             )
         except FileNotFoundError as exc:
-            if self.fs_path == path:
+            if not self.exists:
                 raise self.DoesNotExistError(self) from exc
             if not self.is_dir_checksum:
                 raise
@@ -1398,12 +1400,14 @@ class Output:
 
         assert staging
         assert obj.hash_info
+
+        hardlink = relink and next(iter(self.cache.cache_types), None) == "hardlink"
         with TqdmCallback(desc=f"Adding {self} to cache", unit="file") as cb:
             otransfer(
                 staging,
                 self.cache,
                 {obj.hash_info},
-                hardlink=relink,
+                hardlink=hardlink,
                 shallow=False,
                 callback=cb,
             )
@@ -1421,6 +1425,7 @@ class Output:
                     state=self.repo.state,
                     prompt=prompt.confirm,
                     progress_callback=callback,
+                    old=obj,
                 )
             self.set_exec()
         return obj

dvc/parsing/__init__.py

@@ -563,10 +563,10 @@ class MatrixDefinition:
         iterable = self.resolved_iterable
         assert isinstance(iterable, Mapping)
 
-        ret: dict[str, "DictStrAny"] = {}
+        ret: dict[str, DictStrAny] = {}
         matrix = {key: enumerate(v) for key, v in iterable.items()}
         for combination in product(*matrix.values()):
-            d: "DictStrAny" = {}
+            d: DictStrAny = {}
             fragments: list[str] = []
             for k, (i, v) in zip(matrix.keys(), combination):
                 d[k] = v

dvc/parsing/context.py

@@ -105,7 +105,7 @@ class Meta:
 
     @staticmethod
     def update_path(meta: "Meta", path: Union[str, int]):
-        dpaths = meta.dpaths[:] + [str(path)]
+        dpaths = [*meta.dpaths, str(path)]
         return replace(meta, dpaths=dpaths)
 
     def __str__(self):
@@ -176,8 +176,9 @@ class Container(Node, ABC):
             return value
         if isinstance(value, (list, dict)):
             assert meta
-            container = CtxDict if isinstance(value, dict) else CtxList
-            return container(value, meta=meta)
+            if isinstance(value, dict):
+                return CtxDict(value, meta=meta)
+            return CtxList(value, meta=meta)
         msg = f"Unsupported value of type '{type(value).__name__}' in '{meta}'"
         raise TypeError(msg)
 

dvc/pathspec_math.py

@@ -27,11 +27,9 @@ def _remove_slash(rule):
 
 def _match_all_level(rule):
     if rule[:-1].find("/") >= 0 and not rule.startswith("**/"):
-        if rule.startswith("/"):
-            rule = rule[1:]
+        rule = rule.removeprefix("/")
         return False, rule
-    if rule.startswith("**/"):
-        rule = rule[3:]
+    rule = rule.removeprefix("**/")
     return True, rule
 
 

dvc/progress.py

@@ -41,7 +41,7 @@ class Tqdm(tqdm):
         "miniters": 1,
     }
 
-    def __init__(  # noqa: PLR0913
+    def __init__(
         self,
         iterable=None,
         disable=None,

dvc/render/converter/vega.py

@@ -23,7 +23,7 @@ def _lists(blob: Union[dict, list]) -> Iterable[list]:
     if isinstance(blob, list):
         yield blob
     else:
-        for _, value in blob.items():
+        for value in blob.values():
             if isinstance(value, dict):
                 yield from _lists(value)
             elif isinstance(value, list):

dvc/repo/__init__.py

@@ -147,7 +147,7 @@ class Repo:
     ):
         from dvc.cachemgr import CacheManager
         from dvc.data_cloud import DataCloud
-        from dvc.fs import GitFileSystem, LocalFileSystem, localfs
+        from dvc.fs import GitFileSystem, LocalFileSystem
         from dvc.lock import LockNoop, make_lock
         from dvc.repo.artifacts import Artifacts
         from dvc.repo.datasets import Datasets
@@ -161,7 +161,7 @@ class Repo:
 
         self.url = url
         self._fs_conf = {"repo_factory": repo_factory}
-        self._fs = fs or localfs
+        self._fs = fs or LocalFileSystem()
         self._scm = scm
         self._config = config
         self._remote = remote
@@ -184,12 +184,12 @@ class Repo:
         # used by DVCFileSystem to determine if it should traverse subrepos
         self.subrepos = subrepos
 
-        self.cloud: "DataCloud" = DataCloud(self)
-        self.stage: "StageLoad" = StageLoad(self)
+        self.cloud: DataCloud = DataCloud(self)
+        self.stage: StageLoad = StageLoad(self)
 
-        self.lock: "LockBase"
+        self.lock: LockBase
         self.cache: CacheManager
-        self.state: "StateBase"
+        self.state: StateBase
         if isinstance(self.fs, GitFileSystem) or not self.dvc_dir:
             self.lock = LockNoop()
             self.state = StateNoop()
@@ -206,6 +206,11 @@ class Repo:
                     friendly=True,
                 )
                 os.makedirs(self.site_cache_dir, exist_ok=True)
+                if not fs and (
+                    checksum_jobs := self.config["core"].get("checksum_jobs")
+                ):
+                    self.fs.hash_jobs = checksum_jobs
+
                 self.state = State(self.root_dir, self.site_cache_dir, self.dvcignore)
             else:
                 self.lock = LockNoop()
@@ -604,7 +609,7 @@ class Repo:
         from dvc.fs import GitFileSystem
         from dvc.version import version_tuple
 
-        cache_dir = self.config["core"].get("site_cache_dir") or site_cache_dir()
+        cache_dir = site_cache_dir(self.config["core"].get("site_cache_dir"))
 
         subdir = None
         if isinstance(self.fs, GitFileSystem):

dvc/repo/add.py

@@ -171,11 +171,16 @@ def _add_transfer(
     stage.dump()
 
 
-def _add(stage: "Stage", source: Optional[str] = None, no_commit: bool = False) -> None:
+def _add(
+    stage: "Stage",
+    source: Optional[str] = None,
+    no_commit: bool = False,
+    relink: bool = True,
+) -> None:
     out = stage.outs[0]
     path = out.fs.abspath(source) if source else None
     try:
-        stage.add_outs(path, no_commit=no_commit)
+        stage.add_outs(path, no_commit=no_commit, relink=relink)
     except CacheLinkError:
         stage.dump()
         raise
@@ -194,6 +199,7 @@ def add(
     to_remote: bool = False,
     remote_jobs: Optional[int] = None,
     force: bool = False,
+    relink: bool = True,
 ) -> list["Stage"]:
     add_targets = find_targets(targets, glob=glob)
     if not add_targets:
@@ -224,7 +230,12 @@ def add(
     with warn_link_failures() as link_failures:
         for source, (stage, output_exists) in progress_iter(stages_with_targets):
             try:
-                _add(stage, source if output_exists else None, no_commit=no_commit)
+                _add(
+                    stage,
+                    source if output_exists else None,
+                    no_commit=no_commit,
+                    relink=relink,
+                )
             except CacheLinkError:
                 link_failures.append(stage.relpath)
     return stages

dvc/repo/artifacts.py

@@ -12,7 +12,7 @@ from dvc.exceptions import (
     InvalidArgumentError,
 )
 from dvc.log import logger
-from dvc.utils import relpath, resolve_output
+from dvc.utils import as_posix, relpath, resolve_output
 from dvc.utils.objects import cached_property
 from dvc.utils.serialize import modify_yaml
 
@@ -99,7 +99,7 @@ class Artifacts:
         """Read artifacts from dvc.yaml."""
         artifacts: dict[str, dict[str, Artifact]] = {}
         for dvcfile, dvcfile_artifacts in self.repo.index._artifacts.items():
-            dvcyaml = relpath(dvcfile, self.repo.root_dir)
+            dvcyaml = self.repo.fs.relpath(dvcfile, self.repo.root_dir)
             artifacts[dvcyaml] = {}
             for name, value in dvcfile_artifacts.items():
                 try:
@@ -137,18 +137,16 @@ class Artifacts:
 
         assert not (version and stage)
         name = _reformat_name(name)
-        tags: list["GitTag"] = find(
-            name=name, version=version, stage=stage, scm=self.scm
-        )
+        tags: list[GitTag] = find(name=name, version=version, stage=stage, scm=self.scm)
         if not tags:
             raise ArtifactNotFoundError(name, version=version, stage=stage)
         if version or stage:
             return tags[-1].target
-        gto_tags: list["GTOTag"] = sort_versions(parse_tag(tag) for tag in tags)
+        gto_tags: list[GTOTag] = sort_versions(parse_tag(tag) for tag in tags)
         return gto_tags[0].tag.target
 
-    def get_path(self, name: str):
-        """Return repo fspath for the given artifact."""
+    @classmethod
+    def parse_path(cls, name: str) -> tuple[Optional[str], str]:
         from gto.constants import SEPARATOR_IN_NAME, fullname_re
 
         name = _reformat_name(name)
@@ -158,13 +156,37 @@ class Artifacts:
         dirname = m.group("dirname")
         if dirname:
             dirname = dirname.rstrip(SEPARATOR_IN_NAME)
-        dvcyaml = os.path.join(dirname, PROJECT_FILE) if dirname else PROJECT_FILE
-        artifact_name = m.group("name")
+
+        return dirname, m.group("name")
+
+    def get_path(self, name: str):
+        """Return fspath for the given artifact relative to the git root."""
+        from dvc.fs import GitFileSystem
+
+        dirname, artifact_name = self.parse_path(name)
+        # `name`/`dirname` are expected to be a git root relative.
+        # We convert it to dvc-root relative path so that we can read artifacts
+        # from dvc.yaml file.
+        # But we return dirname intact, as we want to return a git-root relative path.
+        # This is useful when reading from `dvcfs` from remote.
+        fs = self.repo.fs
+        assert self.scm
+        if isinstance(fs, GitFileSystem):
+            scm_root = fs.root_marker
+        else:
+            scm_root = self.scm.root_dir
+
+        dirparts = posixpath.normpath(dirname).split(posixpath.sep) if dirname else ()
+        abspath = fs.join(scm_root, *dirparts, PROJECT_FILE)
+        rela = fs.relpath(abspath, self.repo.root_dir)
         try:
-            artifact = self.read()[dvcyaml][artifact_name]
+            artifact = self.read()[rela][artifact_name]
         except KeyError as exc:
             raise ArtifactNotFoundError(name) from exc
-        return os.path.join(dirname, artifact.path) if dirname else artifact.path
+
+        path = posixpath.join(dirname or "", artifact.path)
+        parts = posixpath.normpath(path).split(posixpath.sep)
+        return os.path.join(*parts)
 
     def download(
         self,
@@ -177,15 +199,28 @@ class Artifacts:
     ) -> tuple[int, str]:
         """Download the specified artifact."""
         from dvc.fs import download as fs_download
+        from dvc.repo import Repo
 
         logger.debug("Trying to download artifact '%s' via DVC", name)
         rev = self.get_rev(name, version=version, stage=stage)
+
+        dirname, _ = self.parse_path(name)
         with self.repo.switch(rev):
-            path = self.get_path(name)
+            root = self.repo.fs.root_marker
+            _dirname = self.repo.fs.join(root, dirname) if dirname else root
+            with Repo(_dirname, fs=self.repo.fs, scm=self.repo.scm) as r:
+                path = r.artifacts.get_path(name)
+                path = self.repo.fs.join(root, as_posix(path))
+                path = self.repo.fs.relpath(path, self.repo.root_dir)
+                # when the `repo` is a subrepo, the path `/subrepo/myart.pkl` for dvcfs
+                # should be translated as `/myart.pkl`,
+                # i.e. relative to the root of the subrepo
+                path = self.repo.fs.join(root, path)
+                path = self.repo.fs.normpath(path)
+
             out = resolve_output(path, out, force=force)
             fs = self.repo.dvcfs
-            fs_path = fs.from_os_path(path)
-            count = fs_download(fs, fs_path, os.path.abspath(out), jobs=jobs)
+            count = len(fs_download(fs, path, os.path.abspath(out), jobs=jobs))
         return count, out
 
     @staticmethod
@@ -230,7 +265,7 @@ class Artifacts:
                 from_infos.append(url)
         except DvcException:
             raise
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             raise DvcException(
                 f"Failed to download artifact '{name}' via Studio"
             ) from exc
@@ -248,7 +283,7 @@ class Artifacts:
         return len(to_infos), relpath(localfs.commonpath(to_infos))
 
     @classmethod
-    def get(  # noqa: PLR0913
+    def get(
         cls,
         url: str,
         name: str,
@@ -277,6 +312,13 @@ class Artifacts:
 
         name = _reformat_name(name)
         saved_exc: Optional[Exception] = None
+
+        local_dvc_studio_config = Config().get("studio", {})
+        args_dvc_studio_config = {}
+        if config and not isinstance(config, dict):
+            config = Config.load_file(config)
+            args_dvc_studio_config = config.get("studio", {})
+
         try:
             logger.trace("Trying studio-only config")
             return cls._download_studio(
@@ -287,14 +329,13 @@ class Artifacts:
                 out=out,
                 force=force,
                 jobs=jobs,
+                dvc_studio_config=local_dvc_studio_config | args_dvc_studio_config,
             )
         except FileExistsLocallyError:
             raise
         except Exception as exc:  # noqa: BLE001
             saved_exc = exc
 
-        if config and not isinstance(config, dict):
-            config = Config.load_file(config)
         with Repo.open(
             url=url,
             subrepos=True,
@@ -304,7 +345,7 @@ class Artifacts:
             remote_config=remote_config,
         ) as repo:
             logger.trace("Trying repo [studio] config")
-            dvc_studio_config = dict(repo.config.get("studio"))
+            repo_dvc_studio_config = repo.config.get("studio", {})
             try:
                 return cls._download_studio(
                     url,
@@ -314,7 +355,9 @@ class Artifacts:
                     out=out,
                     force=force,
                     jobs=jobs,
-                    dvc_studio_config=dvc_studio_config,
+                    dvc_studio_config=local_dvc_studio_config
+                    | repo_dvc_studio_config
+                    | args_dvc_studio_config,
                 )
             except FileExistsLocallyError:
                 raise
@@ -332,7 +375,7 @@ class Artifacts:
                 )
             except FileExistsLocallyError:
                 raise
-            except Exception as exc:  # noqa: BLE001
+            except Exception as exc:
                 if saved_exc:
                     logger.exception(str(saved_exc), exc_info=saved_exc.__cause__)
                 raise DvcException(

dvc/repo/checkout.py

@@ -125,7 +125,7 @@ def checkout(  # noqa: C901
             (StageFileDoesNotExistError, StageFileBadNameError, NoOutputOrStageError),
         ):
             raise CheckoutErrorSuggestGit(target) from exc
-        raise
+        raise  # noqa: PLE0704
 
     view = self.index.targets_view(
         targets, recursive=recursive, with_deps=with_deps, onerror=onerror

dvc/repo/datasets.py

@@ -16,7 +16,7 @@ from dvc.types import StrPath
 from dvc_data.hashfile.meta import Meta
 
 if TYPE_CHECKING:
-    from dvcx.dataset import DatasetRecord, DatasetVersion  # type: ignore[import]
+    from datachain.dataset import DatasetRecord, DatasetVersion  # type: ignore[import]
     from typing_extensions import Self
 
     from dvc.repo import Repo
@@ -29,10 +29,10 @@ def _get_dataset_record(name: str) -> "DatasetRecord":
     from dvc.exceptions import DvcException
 
     try:
-        from dvcx.catalog import get_catalog  # type: ignore[import]
+        from datachain.catalog import get_catalog  # type: ignore[import]
 
     except ImportError as exc:
-        raise DvcException("dvcx is not installed") from exc
+        raise DvcException("datachain is not installed") from exc
 
     catalog = get_catalog()
     return catalog.get_remote_dataset(name)
@@ -76,7 +76,7 @@ class SerDe:
         return asdict(self, filter=filter_defaults, value_serializer=value_serializer)
 
     @classmethod
-    def from_dict(cls: type["Self"], d: dict[str, Any]) -> "Self":
+    def from_dict(cls, d: dict[str, Any]) -> "Self":
         _fields = fields(cast("type[AttrsInstance]", cls))
         kwargs = {f.name: d[f.name] for f in _fields if f.name in d}
         return cls(**kwargs)
@@ -86,7 +86,7 @@ class SerDe:
 class DatasetSpec(SerDe):
     name: str
     url: str
-    type: Literal["dvc", "dvcx", "url"]
+    type: Literal["dvc", "dc", "url"]
 
 
 @frozen(kw_only=True)
@@ -108,7 +108,7 @@ class DVCDatasetLock(DVCDatasetSpec):
 
 
 @frozen(kw_only=True)
-class DVCXDatasetLock(DatasetSpec):
+class DatachainDatasetLock(DatasetSpec):
     version: int
     created_at: datetime = field(converter=to_datetime)
 
@@ -160,13 +160,13 @@ class DVCDataset:
 
 
 @frozen(kw_only=True)
-class DVCXDataset:
+class DatachainDataset:
     manifest_path: str
     spec: "DatasetSpec"
-    lock: "Optional[DVCXDatasetLock]" = field(default=None)
+    lock: "Optional[DatachainDatasetLock]" = field(default=None)
     _invalidated: bool = field(default=False, eq=False, repr=False)
 
-    type: ClassVar[Literal["dvcx"]] = "dvcx"
+    type: ClassVar[Literal["dc"]] = "dc"
 
     @property
     def pinned(self) -> bool:
@@ -193,7 +193,7 @@ class DVCXDataset:
         name, _version = self.name_version
         version = version if version is not None else _version
         version_info = _get_dataset_info(name, record=record, version=version)
-        lock = DVCXDatasetLock(
+        lock = DatachainDatasetLock(
             **self.spec.to_dict(),
             version=version_info.version,
             created_at=version_info.created_at,
@@ -226,9 +226,9 @@ class URLDataset:
         return evolve(self, lock=lock)
 
 
-Lock = Union[DVCDatasetLock, DVCXDatasetLock, URLDatasetLock]
+Lock = Union[DVCDatasetLock, DatachainDatasetLock, URLDatasetLock]
 Spec = Union[DatasetSpec, DVCDatasetSpec]
-Dataset = Union[DVCDataset, DVCXDataset, URLDataset]
+Dataset = Union[DVCDataset, DatachainDataset, URLDataset]
 
 
 class DatasetNotFoundError(DvcException, KeyError):
@@ -242,7 +242,7 @@ class DatasetNotFoundError(DvcException, KeyError):
 
 class Datasets(Mapping[str, Dataset]):
     def __init__(self, repo: "Repo") -> None:
-        self.repo: "Repo" = repo
+        self.repo: Repo = repo
 
     def __repr__(self):
         return repr(dict(self))
@@ -307,13 +307,13 @@ class Datasets(Mapping[str, Dataset]):
             raise ValueError("type should be present in spec")
         if typ == "dvc":
             return DVCDatasetSpec.from_dict(spec)
-        if typ in {"dvcx", "url"}:
+        if typ in {"dc", "url"}:
             return DatasetSpec.from_dict(spec)
         raise ValueError(f"unknown dataset type: {spec.get('type', '')}")
 
     @staticmethod
     def _lock_from_info(lock: Optional[dict[str, Any]]) -> Optional[Lock]:
-        kl = {"dvc": DVCDatasetLock, "dvcx": DVCXDatasetLock, "url": URLDatasetLock}
+        kl = {"dvc": DVCDatasetLock, "dc": DatachainDatasetLock, "url": URLDatasetLock}
         if lock and (cls := kl.get(lock.get("type", ""))):  # type: ignore[assignment]
             return cls.from_dict(lock)  # type: ignore[attr-defined]
         return None
@@ -356,9 +356,9 @@ class Datasets(Mapping[str, Dataset]):
                 lock=lock,
                 invalidated=_invalidated,
             )
-        if spec.type == "dvcx":
-            assert lock is None or isinstance(lock, DVCXDatasetLock)
-            return DVCXDataset(
+        if spec.type == "dc":
+            assert lock is None or isinstance(lock, DatachainDatasetLock)
+            return DatachainDataset(
                 manifest_path=manifest_path,
                 spec=spec,
                 lock=lock,
@@ -374,7 +374,7 @@ class Datasets(Mapping[str, Dataset]):
         manifest_path: StrPath = "dvc.yaml",
         **kwargs: Any,
     ) -> Dataset:
-        assert type in {"dvc", "dvcx", "url"}
+        assert type in {"dvc", "dc", "url"}
         kwargs.update({"name": name, "url": url, "type": type})
         dataset = self._build_dataset(os.path.abspath(manifest_path), kwargs)
         dataset = dataset.update(self.repo)
@@ -389,13 +389,16 @@ class Datasets(Mapping[str, Dataset]):
 
         if dataset.type == "url" and (version or kwargs.get("rev")):
             raise ValueError("cannot update version/revision for a url")
-        if dataset.type == "dvcx" and version is not None:
+        if dataset.type == "dc" and version is not None:
             if not isinstance(version, int):
                 raise TypeError(
-                    f"dvcx version has to be an integer, got {type(version).__name__!r}"
+                    "DataChain dataset version has to be an integer, "
+                    f"got {type(version).__name__!r}"
                 )
             if version < 1:
-                raise ValueError(f"dvcx version should be >=1, got {version}")
+                raise ValueError(
+                    f"DataChain dataset version should be >=1, got {version}"
+                )
 
         new = dataset.update(self.repo, **kwargs)
 

dvc/repo/diff.py

@@ -26,7 +26,7 @@ def _diff(old, new, data_keys, with_missing=False):
     from dvc_data.index.diff import ADD, DELETE, MODIFY, RENAME
     from dvc_data.index.diff import diff as idiff
 
-    ret: "dict[str, list[dict]]" = {
+    ret: dict[str, list[dict]] = {
         "added": [],
         "deleted": [],
         "modified": [],

dvc/repo/du.py

@@ -6,7 +6,7 @@ def du(
     path: Optional[str] = None,
     rev: Optional[str] = None,
     summarize: bool = False,
-    config: Union[None, dict[str, Any], str] = None,
+    config: Union[dict[str, Any], str, None] = None,
     remote: Optional[str] = None,
     remote_config: Optional[dict] = None,
 ):

dvc/repo/experiments/__init__.py

@@ -119,7 +119,7 @@ class Experiments:
         **kwargs,
     ):
         """Reproduce and checkout a single (standalone) experiment."""
-        exp_queue: "BaseStashQueue" = (
+        exp_queue: BaseStashQueue = (
             self.tempdir_queue if tmp_dir else self.workspace_queue
         )
         self.queue_one(exp_queue, **kwargs)
@@ -205,7 +205,7 @@ class Experiments:
         Experiment will be derived from the current workspace.
         """
 
-        name = kwargs.get("name", None)
+        name = kwargs.get("name")
         baseline_sha = kwargs.get("baseline_rev") or self.repo.scm.get_rev()
 
         if name:
@@ -249,7 +249,7 @@ class Experiments:
         exec_results = queue.reproduce(copy_paths=copy_paths, message=message)
 
         results: dict[str, str] = {}
-        for _, exp_result in exec_results.items():
+        for exp_result in exec_results.values():
             results.update(exp_result)
         return results
 

dvc/repo/experiments/apply.py

@@ -25,7 +25,7 @@ def apply(repo: "Repo", rev: str, **kwargs):
     from dvc.repo.checkout import checkout as dvc_checkout
     from dvc.scm import RevError, resolve_rev
 
-    exps: "Experiments" = repo.experiments
+    exps: Experiments = repo.experiments
 
     is_stash: bool = False
 
@@ -57,7 +57,7 @@ def apply(repo: "Repo", rev: str, **kwargs):
 
 
 def _apply(repo: "Repo", rev: str, name: Optional[str] = None, is_stash: bool = False):
-    exps: "Experiments" = repo.experiments
+    exps: Experiments = repo.experiments
 
     with exps.apply_stash.preserve_workspace(rev, name=name):
         with repo.scm.detach_head(rev, force=True):

dvc/repo/experiments/collect.py

@@ -69,11 +69,11 @@ def collect_rev(
 
 def _collect_rev(
     repo: "Repo",
-    rev: str,
+    revision: str,
     param_deps: bool = False,
     **kwargs,
 ) -> SerializableExp:
-    with repo.switch(rev) as rev:
+    with repo.switch(revision) as rev:
         if rev == "workspace":
             timestamp: Optional[datetime] = None
         else:
@@ -174,7 +174,7 @@ def collect_active(
     """
     if not baseline_revs:
         return {}
-    result: dict[str, list["ExpRange"]] = {}
+    result: dict[str, list[ExpRange]] = {}
     exps = repo.experiments
     for queue in (exps.workspace_queue, exps.tempdir_queue, exps.celery_queue):
         for baseline, active_exps in queue.collect_active_data(
@@ -220,7 +220,7 @@ def collect_successful(
     Returns:
         Dict mapping baseline revision to successful experiments.
     """
-    result: dict[str, list["ExpRange"]] = {}
+    result: dict[str, list[ExpRange]] = {}
     for baseline_rev in baseline_revs:
         result[baseline_rev] = list(_collect_baseline(repo, baseline_rev, **kwargs))
     return result
@@ -304,7 +304,7 @@ def collect(
         )
 
     workspace_data = collect_rev(repo, "workspace", **kwargs)
-    result: list["ExpState"] = [workspace_data]
+    result: list[ExpState] = [workspace_data]
     queued = collect_queued(repo, baseline_revs, **kwargs) if not hide_queued else {}
     active = collect_active(repo, baseline_revs, **kwargs)
     failed = collect_failed(repo, baseline_revs, **kwargs) if not hide_failed else {}

dvc/repo/experiments/executor/base.py

@@ -199,7 +199,7 @@ class BaseExecutor(ABC):
     @classmethod
     def from_info(cls, info: "ExecutorInfo") -> "Self":
         if info.result_hash:
-            result: Optional["ExecutorResult"] = ExecutorResult(
+            result: Optional[ExecutorResult] = ExecutorResult(
                 info.result_hash,
                 (ExpRefInfo.from_ref(info.result_ref) if info.result_ref else None),
                 info.result_force,
@@ -288,7 +288,7 @@ class BaseExecutor(ABC):
             stages = []
             if targets:
                 for target in targets:
-                    stages.append(
+                    stages.append(  # noqa: PERF401
                         dvc.commit(
                             target, recursive=recursive, force=True, relink=False
                         )
@@ -470,7 +470,7 @@ class BaseExecutor(ABC):
             cls._set_log_level(log_level)
 
         exp_hash: Optional[str] = None
-        exp_ref: Optional["ExpRefInfo"] = None
+        exp_ref: Optional[ExpRefInfo] = None
         repro_force: bool = False
 
         if info.name:
@@ -559,7 +559,7 @@ class BaseExecutor(ABC):
             )
 
         ref: Optional[str] = dvc.scm.get_ref(EXEC_BRANCH, follow=False)
-        exp_ref: Optional["ExpRefInfo"] = ExpRefInfo.from_ref(ref) if ref else None
+        exp_ref: Optional[ExpRefInfo] = ExpRefInfo.from_ref(ref) if ref else None
         if cls.WARN_UNTRACKED:
             untracked = dvc.scm.untracked_files()
             if untracked:

dvc/repo/experiments/executor/local.py

@@ -201,7 +201,7 @@ class WorkspaceExecutor(BaseLocalExecutor):
     @classmethod
     def from_stash_entry(cls, repo: "Repo", entry: "ExpStashEntry", **kwargs):
         root_dir = repo.scm.root_dir
-        executor: "WorkspaceExecutor" = cls._from_stash_entry(
+        executor: WorkspaceExecutor = cls._from_stash_entry(
             repo, entry, root_dir, **kwargs
         )
         logger.debug("Init workspace executor in '%s'", root_dir)

dvc/repo/experiments/pull.py

@@ -30,7 +30,7 @@ def pull(  # noqa: C901
     pull_cache: bool = False,
     **kwargs,
 ) -> Iterable[str]:
-    exp_ref_set: set["ExpRefInfo"] = set()
+    exp_ref_set: set[ExpRefInfo] = set()
     if all_commits:
         exp_ref_set.update(exp_refs(repo.scm, git_remote))
     elif exp_names:
@@ -55,7 +55,7 @@ def pull(  # noqa: C901
         rev_dict = iter_revs(repo.scm, rev, num)
         rev_set = set(rev_dict.keys())
         ref_info_dict = exp_refs_by_baseline(repo.scm, rev_set, git_remote)
-        for _, ref_info_list in ref_info_dict.items():
+        for ref_info_list in ref_info_dict.values():
             exp_ref_set.update(ref_info_list)
 
     pull_result = _pull(repo, git_remote, exp_ref_set, force)
@@ -97,7 +97,7 @@ def _pull(
     def group_result(refspec):
         return results[str(refspec)]
 
-    pull_result: Mapping[SyncStatus, list["ExpRefInfo"]] = group_by(group_result, refs)
+    pull_result: Mapping[SyncStatus, list[ExpRefInfo]] = group_by(group_result, refs)
 
     return pull_result
 

dvc/repo/experiments/push.py

@@ -78,7 +78,7 @@ def exp_refs_from_rev(scm: "Git", rev: list[str], num: int = 1) -> set["ExpRefIn
     rev_dict = iter_revs(scm, rev, num)
     rev_set = set(rev_dict.keys())
     ref_info_dict = exp_refs_by_baseline(scm, rev_set)
-    for _, ref_info_list in ref_info_dict.items():
+    for ref_info_list in ref_info_dict.values():
         exp_ref_set.update(ref_info_list)
     return exp_ref_set
 
@@ -96,7 +96,7 @@ def push(
     push_cache: bool = False,
     **kwargs: Any,
 ) -> dict[str, Any]:
-    exp_ref_set: set["ExpRefInfo"] = set()
+    exp_ref_set: set[ExpRefInfo] = set()
     assert isinstance(repo.scm, Git)
     if all_commits:
         exp_ref_set.update(exp_refs(repo.scm))
@@ -162,7 +162,7 @@ def _push(
     def group_result(refspec):
         return results[str(refspec)]
 
-    pull_result: Mapping[SyncStatus, list["ExpRefInfo"]] = group_by(group_result, refs)
+    pull_result: Mapping[SyncStatus, list[ExpRefInfo]] = group_by(group_result, refs)
 
     return pull_result
 

dvc/repo/experiments/queue/base.py

@@ -372,7 +372,7 @@ class BaseStashQueue(ABC):
         )
 
     def _stash_commit_deps(self, *args, **kwargs):
-        if len(args):
+        if args:
             targets = args[0]
         else:
             targets = kwargs.get("targets")
@@ -609,9 +609,9 @@ class BaseStashQueue(ABC):
                 result[exp_name] = entry_name_dict[exp_name]
                 continue
             if self.scm.is_sha(exp_name):
-                for rev in entry_rev_dict:
+                for rev, entry in entry_rev_dict.items():
                     if rev.startswith(exp_name.lower()):
-                        result[exp_name] = entry_rev_dict[rev]
+                        result[exp_name] = entry
                         break
 
         return result

dvc/repo/experiments/queue/celery.py

@@ -323,7 +323,7 @@ class LocalCeleryQueue(BaseStashQueue):
     def _get_running_task_ids(self) -> set[str]:
         running_task_ids: set[str] = set()
         active_workers = self.worker_status()
-        for _, tasks in active_workers.items():
+        for tasks in active_workers.values():
             task = first(tasks)
             if task:
                 running_task_ids.add(task["id"])
@@ -470,11 +470,11 @@ class LocalCeleryQueue(BaseStashQueue):
             exp_names = [exp_names]
         results: dict[str, ExpRefAndQueueEntry] = {}
 
-        exp_ref_match: dict[str, Optional["ExpRefInfo"]] = resolve_name(
+        exp_ref_match: dict[str, Optional[ExpRefInfo]] = resolve_name(
             self.scm, exp_names, git_remote
         )
         if not git_remote:
-            queue_entry_match: dict[str, Optional["QueueEntry"]] = (
+            queue_entry_match: dict[str, Optional[QueueEntry]] = (
                 self.match_queue_entry_by_name(
                     exp_names, self.iter_queued(), self.iter_done()
                 )
@@ -613,7 +613,7 @@ class LocalCeleryQueue(BaseStashQueue):
         """Map exp refs to any available successful executors."""
         from dvc.repo.experiments.serialize import ExpExecutor, LocalExpExecutor
 
-        result: dict[str, "ExpExecutor"] = {}
+        result: dict[str, ExpExecutor] = {}
         for entry, exec_result in self.iter_success():
             if baseline_revs and entry.baseline_rev not in baseline_revs:
                 continue

dvc/repo/experiments/queue/remove.py

@@ -21,11 +21,11 @@ def remove_tasks(  # noqa: C901, PLR0912
     """
     from celery.result import AsyncResult
 
-    stash_revs: dict[str, "ExpStashEntry"] = {}
-    failed_stash_revs: list["ExpStashEntry"] = []
-    done_entry_set: set["QueueEntry"] = set()
+    stash_revs: dict[str, ExpStashEntry] = {}
+    failed_stash_revs: list[ExpStashEntry] = []
+    done_entry_set: set[QueueEntry] = set()
     stash_rev_all = celery_queue.stash.stash_revs
-    failed_rev_all: dict[str, "ExpStashEntry"] = {}
+    failed_rev_all: dict[str, ExpStashEntry] = {}
     if celery_queue.failed_stash:
         failed_rev_all = celery_queue.failed_stash.stash_revs
     for entry in queue_entries:
@@ -90,17 +90,17 @@ def celery_clear(
     """
 
     removed: list[str] = []
-    entry_list: list["QueueEntry"] = []
+    entry_list: list[QueueEntry] = []
     if queued:
-        queue_entries: list["QueueEntry"] = list(self.iter_queued())
+        queue_entries: list[QueueEntry] = list(self.iter_queued())
         entry_list.extend(queue_entries)
         removed.extend(_get_names(queue_entries))
     if failed:
-        failed_tasks: list["QueueDoneResult"] = list(self.iter_failed())
+        failed_tasks: list[QueueDoneResult] = list(self.iter_failed())
         entry_list.extend([result.entry for result in failed_tasks])
         removed.extend(_get_names(failed_tasks))
     if success:
-        success_tasks: list["QueueDoneResult"] = list(self.iter_success())
+        success_tasks: list[QueueDoneResult] = list(self.iter_success())
         entry_list.extend([result.entry for result in success_tasks])
         removed.extend(_get_names(success_tasks))
 
@@ -125,7 +125,7 @@ def celery_remove(self: "LocalCeleryQueue", revs: Collection[str]) -> list[str]:
 
     remained: list[str] = []
     removed: list[str] = []
-    entry_to_remove: list["QueueEntry"] = []
+    entry_to_remove: list[QueueEntry] = []
     for name, entry in match_results.items():
         if entry:
             entry_to_remove.append(entry)

dvc/repo/experiments/queue/tempdir.py

@@ -125,7 +125,7 @@ class TempDirQueue(WorkspaceQueue):
                 )
         except DvcException:
             raise
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             raise DvcException(f"Failed to reproduce experiment '{rev[:7]}'") from exc
         finally:
             executor.cleanup(infofile)

dvc/repo/experiments/queue/utils.py

@@ -25,7 +25,7 @@ def get_remote_executor_refs(scm: "Git", remote_url: str) -> list[str]:
     refs = []
     for ref in iter_remote_refs(scm, remote_url, base=EXPS_NAMESPACE):
         if not ref.startswith(EXEC_NAMESPACE) and ref != EXPS_STASH:
-            refs.append(ref)
+            refs.append(ref)  # noqa: PERF401
     return refs
 
 

dvc/repo/experiments/queue/workspace.py

@@ -131,7 +131,7 @@ class WorkspaceQueue(BaseStashQueue):
                 )
         except DvcException:
             raise
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             raise DvcException(f"Failed to reproduce experiment '{rev[:7]}'") from exc
         finally:
             executor.cleanup(infofile)

dvc/repo/experiments/refs.py

@@ -67,3 +67,12 @@ class ExpRefInfo:
         baseline_sha = parts[2] + parts[3]
         name = parts[4] if len(parts) == 5 else None
         return cls(baseline_sha, name)
+
+    def __eq__(self, other):
+        if not isinstance(other, ExpRefInfo):
+            return False
+
+        return self.baseline_sha == other.baseline_sha and self.name == other.name
+
+    def __hash__(self):
+        return hash((self.baseline_sha, self.name))

dvc/repo/experiments/remove.py

@@ -6,7 +6,7 @@ from dvc.repo import locked
 from dvc.repo.scm_context import scm_context
 from dvc.scm import Git, iter_revs
 
-from .exceptions import UnresolvedExpNamesError
+from .exceptions import InvalidArgumentError, UnresolvedExpNamesError
 from .utils import exp_refs, exp_refs_by_baseline, push_refspec
 
 if TYPE_CHECKING:
@@ -24,27 +24,34 @@ logger = logger.getChild(__name__)
 @scm_context
 def remove(  # noqa: C901, PLR0912
     repo: "Repo",
-    exp_names: Union[None, str, list[str]] = None,
+    exp_names: Union[str, list[str], None] = None,
     rev: Optional[Union[list[str], str]] = None,
     all_commits: bool = False,
     num: int = 1,
     queue: bool = False,
     git_remote: Optional[str] = None,
+    keep: bool = False,
 ) -> list[str]:
     removed: list[str] = []
+
+    if all([keep, queue]):
+        raise InvalidArgumentError("Cannot use both `--keep` and `--queue`.")
+
     if not any([exp_names, queue, all_commits, rev]):
         return removed
-    celery_queue: "LocalCeleryQueue" = repo.experiments.celery_queue
+
+    celery_queue: LocalCeleryQueue = repo.experiments.celery_queue
 
     if queue:
         removed.extend(celery_queue.clear(queued=True))
 
     assert isinstance(repo.scm, Git)
 
-    exp_ref_list: list["ExpRefInfo"] = []
-    queue_entry_list: list["QueueEntry"] = []
+    exp_ref_list: list[ExpRefInfo] = []
+    queue_entry_list: list[QueueEntry] = []
+
     if exp_names:
-        results: dict[str, "ExpRefAndQueueEntry"] = (
+        results: dict[str, ExpRefAndQueueEntry] = (
             celery_queue.get_ref_and_entry_by_names(exp_names, git_remote)
         )
         remained: list[str] = []
@@ -68,6 +75,10 @@ def remove(  # noqa: C901, PLR0912
         exp_ref_list.extend(exp_ref_dict.values())
     elif all_commits:
         exp_ref_list.extend(exp_refs(repo.scm, git_remote))
+        removed.extend([ref.name for ref in exp_ref_list])
+
+    if keep:
+        exp_ref_list = list(set(exp_refs(repo.scm, git_remote)) - set(exp_ref_list))
         removed = [ref.name for ref in exp_ref_list]
 
     if exp_ref_list:
@@ -83,6 +94,7 @@ def remove(  # noqa: C901, PLR0912
 
         removed_refs = [str(r) for r in exp_ref_list]
         notify_refs_to_studio(repo, git_remote, removed=removed_refs)
+
     return removed
 
 
@@ -94,11 +106,11 @@ def _resolve_exp_by_baseline(
 ) -> dict[str, "ExpRefInfo"]:
     assert isinstance(repo.scm, Git)
 
-    commit_ref_dict: dict[str, "ExpRefInfo"] = {}
+    commit_ref_dict: dict[str, ExpRefInfo] = {}
     rev_dict = iter_revs(repo.scm, rev, num)
     rev_set = set(rev_dict.keys())
     ref_info_dict = exp_refs_by_baseline(repo.scm, rev_set, git_remote)
-    for _, ref_info_list in ref_info_dict.items():
+    for ref_info_list in ref_info_dict.values():
         for ref_info in ref_info_list:
             commit_ref_dict[ref_info.name] = ref_info
     return commit_ref_dict

dvc/repo/experiments/run.py

@@ -93,7 +93,7 @@ def run(  # noqa: C901, PLR0912
 
     for idx, sweep_overrides in enumerate(sweeps):
         if hydra_sweep and name_prefix is not None:
-            kwargs["name"] = f"{name_prefix}-{idx+1}"
+            kwargs["name"] = f"{name_prefix}-{idx + 1}"
         queue_entry = repo.experiments.queue_one(
             repo.experiments.celery_queue,
             targets=targets,

dvc/repo/experiments/serialize.py

@@ -50,7 +50,12 @@ class SerializableExp:
         Params, metrics, deps, outs are filled via repo fs/index, all other fields
         should be passed via kwargs.
         """
-        from dvc.dependency import ParamsDependency, RepoDependency
+        from dvc.dependency import (
+            DatasetDependency,
+            DbDependency,
+            ParamsDependency,
+            RepoDependency,
+        )
 
         rev = rev or repo.get_rev()
         assert rev
@@ -68,7 +73,10 @@ class SerializableExp:
                     nfiles=dep.meta.nfiles if dep.meta else None,
                 )
                 for dep in repo.index.deps
-                if not isinstance(dep, (ParamsDependency, RepoDependency))
+                if not isinstance(
+                    dep,
+                    (ParamsDependency, RepoDependency, DatasetDependency, DbDependency),
+                )
             },
             outs={
                 relpath(out.fs_path, repo.root_dir): ExpOut(

dvc/repo/experiments/show.py

@@ -117,7 +117,7 @@ def _build_rows(
     **kwargs,
 ) -> Iterator[tuple["CellT", ...]]:
     for baseline in baseline_states:
-        row: dict[str, "CellT"] = dict.fromkeys(all_headers, fill_value)
+        row: dict[str, CellT] = dict.fromkeys(all_headers, fill_value)
         row["Experiment"] = ""
         if baseline.name:
             row["rev"] = baseline.name
@@ -232,7 +232,7 @@ def _exp_range_rows(
         logger.debug("Returning tip commit for legacy checkpoint exp")
     exp = first(exp_range.revs)
     if exp:
-        row: dict[str, "CellT"] = dict.fromkeys(all_headers, fill_value)
+        row: dict[str, CellT] = dict.fromkeys(all_headers, fill_value)
         row["Experiment"] = exp.name or ""
         row["rev"] = exp.rev[:7] if Git.is_sha(exp.rev) else exp.rev
         row["typ"] = "branch_base" if is_base else "branch_commit"

dvc/repo/experiments/utils.py

@@ -199,7 +199,7 @@ def exp_commits(
     scm: "Git", ref_infos: Optional[Iterable[ExpRefInfo]] = None
 ) -> Iterable[str]:
     """Iterate over all experiment commits."""
-    shas: set["str"] = set()
+    shas: set[str] = set()
     refs = ref_infos if ref_infos else exp_refs(scm)
     for ref_info in refs:
         shas.update(scm.branch_revs(str(ref_info), ref_info.baseline_sha))

dvc/repo/fetch.py

@@ -2,6 +2,7 @@ from typing import TYPE_CHECKING
 
 from dvc.exceptions import DownloadError
 from dvc.log import logger
+from dvc.stage.cache import RunCacheNotSupported
 from dvc.ui import ui
 from dvc_data.index import DataIndex, FileStorage
 
@@ -136,6 +137,8 @@ def fetch(  # noqa: PLR0913
     try:
         if run_cache:
             self.stage_cache.pull(remote)
+    except RunCacheNotSupported as e:
+        logger.debug("failed to pull run cache: %s", e)
     except DownloadError as exc:
         failed_count += exc.amount
 
@@ -200,7 +203,7 @@ def fetch(  # noqa: PLR0913
 
 
 def _log_unversioned(data: list["DataIndex"]) -> tuple[list["DataIndex"], int]:
-    ret: list["DataIndex"] = []
+    ret: list[DataIndex] = []
     unversioned: list[str] = []
     for fs_index in data:
         remote = fs_index.storage_map[()].remote

dvc/repo/gc.py

@@ -50,7 +50,7 @@ def _used_obj_ids_not_in_remote(
 
 
 @locked
-def gc(  # noqa: PLR0913, C901
+def gc(  # noqa: C901, PLR0912, PLR0913
     self: "Repo",
     all_branches: bool = False,
     cloud: bool = False,
@@ -96,7 +96,7 @@ def gc(  # noqa: PLR0913, C901
         repos = []
     all_repos = [Repo(path) for path in repos]
 
-    odb_to_obj_ids: "ObjectContainer" = {}
+    odb_to_obj_ids: ObjectContainer = {}
     with ExitStack() as stack:
         for repo in all_repos:
             stack.enter_context(repo.lock)

dvc/repo/get.py

@@ -54,7 +54,7 @@ def get(
         from dvc.fs import download
         from dvc.fs.data import DataFileSystem
 
-        fs: Union[DataFileSystem, "DVCFileSystem"]
+        fs: Union[DataFileSystem, DVCFileSystem]
         if os.path.isabs(path):
             fs = DataFileSystem(index=repo.index.data["local"])
             fs_path = fs.from_os_path(path)

dvc/repo/graph.py

@@ -22,7 +22,7 @@ def check_acyclic(graph: "DiGraph") -> None:
     except nx.NetworkXNoCycle:
         return
 
-    stages: set["Stage"] = set()
+    stages: set[Stage] = set()
     for from_node, to_node, _ in edges:
         stages.add(from_node)
         stages.add(to_node)

dvc/repo/imp_db.py

@@ -4,7 +4,6 @@ from funcy import compact
 
 from dvc.exceptions import OutputDuplicationError
 from dvc.repo.scm_context import scm_context
-from dvc.ui import ui
 from dvc.utils import resolve_output, resolve_paths
 
 if TYPE_CHECKING:
@@ -25,11 +24,6 @@ def imp_db(
     force: bool = False,
     connection: Optional[str] = None,
 ):
-    ui.warn("WARNING: import-db is an experimental feature.")
-    ui.warn(
-        "The functionality may change or break without notice, "
-        "which could lead to unexpected behavior."
-    )
     assert sql or table
     assert output_format in ("csv", "json")
 

dvc/repo/index.py

@@ -300,7 +300,7 @@ class Index:
         self._artifacts = artifacts or {}
         self._datasets: dict[str, list[dict[str, Any]]] = datasets or {}
         self._datasets_lock: dict[str, list[dict[str, Any]]] = datasets_lock or {}
-        self._collected_targets: dict[int, list["StageInfo"]] = {}
+        self._collected_targets: dict[int, list[StageInfo]] = {}
 
     @cached_property
     def rev(self) -> Optional[str]:
@@ -416,7 +416,7 @@ class Index:
 
     @cached_property
     def out_data_keys(self) -> dict[str, set["DataIndexKey"]]:
-        by_workspace: dict[str, set["DataIndexKey"]] = defaultdict(set)
+        by_workspace: dict[str, set[DataIndexKey]] = defaultdict(set)
 
         by_workspace["repo"] = set()
         by_workspace["local"] = set()
@@ -470,7 +470,7 @@ class Index:
 
     @cached_property
     def data_keys(self) -> dict[str, set["DataIndexKey"]]:
-        by_workspace: dict[str, set["DataIndexKey"]] = defaultdict(set)
+        by_workspace: dict[str, set[DataIndexKey]] = defaultdict(set)
 
         by_workspace["repo"] = set()
         by_workspace["local"] = set()
@@ -488,7 +488,7 @@ class Index:
     def metric_keys(self) -> dict[str, set["DataIndexKey"]]:
         from .metrics.show import _collect_top_level_metrics
 
-        by_workspace: dict[str, set["DataIndexKey"]] = defaultdict(set)
+        by_workspace: dict[str, set[DataIndexKey]] = defaultdict(set)
 
         by_workspace["repo"] = set()
 
@@ -509,7 +509,7 @@ class Index:
     def param_keys(self) -> dict[str, set["DataIndexKey"]]:
         from .params.show import _collect_top_level_params
 
-        by_workspace: dict[str, set["DataIndexKey"]] = defaultdict(set)
+        by_workspace: dict[str, set[DataIndexKey]] = defaultdict(set)
         by_workspace["repo"] = set()
 
         param_paths = _collect_top_level_params(self.repo)
@@ -525,7 +525,7 @@ class Index:
 
     @cached_property
     def plot_keys(self) -> dict[str, set["DataIndexKey"]]:
-        by_workspace: dict[str, set["DataIndexKey"]] = defaultdict(set)
+        by_workspace: dict[str, set[DataIndexKey]] = defaultdict(set)
 
         by_workspace["repo"] = set()
 
@@ -548,7 +548,7 @@ class Index:
 
     @cached_property
     def data(self) -> "dict[str, DataIndex]":
-        prefix: "DataIndexKey"
+        prefix: DataIndexKey
         loaded = False
 
         index = self.repo.data_index
@@ -600,7 +600,7 @@ class Index:
         if not onerror:
 
             def onerror(_target, _exc):
-                raise
+                raise  # noqa: PLE0704
 
         targets = ensure_list(targets)
         if not targets:
@@ -627,7 +627,7 @@ class Index:
         jobs: Optional[int] = None,
         push: bool = False,
     ) -> "ObjectContainer":
-        used: "ObjectContainer" = defaultdict(set)
+        used: ObjectContainer = defaultdict(set)
         pairs = self.collect_targets(targets, recursive=recursive, with_deps=with_deps)
         for stage, filter_info in pairs:
             for odb, objs in stage.get_used_objs(
@@ -748,7 +748,7 @@ class IndexView:
 
     @cached_property
     def out_data_keys(self) -> dict[str, set["DataIndexKey"]]:
-        by_workspace: dict[str, set["DataIndexKey"]] = defaultdict(set)
+        by_workspace: dict[str, set[DataIndexKey]] = defaultdict(set)
 
         by_workspace["repo"] = set()
         by_workspace["local"] = set()
@@ -764,7 +764,7 @@ class IndexView:
 
     @cached_property
     def _data_prefixes(self) -> dict[str, "_DataPrefixes"]:
-        prefixes: dict[str, "_DataPrefixes"] = defaultdict(
+        prefixes: dict[str, _DataPrefixes] = defaultdict(
             lambda: _DataPrefixes(set(), set())
         )
         for out, filter_info in self._filtered_outs:
@@ -781,7 +781,7 @@ class IndexView:
 
     @cached_property
     def data_keys(self) -> dict[str, set["DataIndexKey"]]:
-        ret: dict[str, set["DataIndexKey"]] = defaultdict(set)
+        ret: dict[str, set[DataIndexKey]] = defaultdict(set)
 
         for out, filter_info in self._filtered_outs:
             if not out.use_cache:
@@ -811,7 +811,7 @@ class IndexView:
             except KeyError:
                 return False
 
-        data: dict[str, Union["DataIndex", "DataIndexView"]] = {}
+        data: dict[str, Union[DataIndex, DataIndexView]] = {}
         for workspace, data_index in self._index.data.items():
             if self.stages:
                 data[workspace] = view(data_index, partial(key_filter, workspace))

dvc/repo/install.py

@@ -1,5 +1,6 @@
 from typing import TYPE_CHECKING
 
+from dvc import version_tuple
 from dvc.exceptions import DvcException
 
 if TYPE_CHECKING:
@@ -16,19 +17,19 @@ def pre_commit_install(scm: "Git") -> None:
     with modify_yaml(config_path) as config:
         entry = {
             "repo": "https://github.com/iterative/dvc",
-            "rev": "main",
+            "rev": ".".join(map(str, version_tuple[:3])),
             "hooks": [
                 {
                     "id": "dvc-pre-commit",
                     "additional_dependencies": [".[all]"],
                     "language_version": "python3",
-                    "stages": ["commit"],
+                    "stages": ["pre-commit"],
                 },
                 {
                     "id": "dvc-pre-push",
                     "additional_dependencies": [".[all]"],
                     "language_version": "python3",
-                    "stages": ["push"],
+                    "stages": ["pre-push"],
                 },
                 {
                     "id": "dvc-post-checkout",

dvc/repo/ls.py

@@ -4,8 +4,44 @@ from typing import TYPE_CHECKING, Any, Optional, Union
 if TYPE_CHECKING:
     from dvc.fs.dvc import DVCFileSystem
 
+
+def _open_repo(
+    url: str,
+    rev: Optional[str] = None,
+    config: Union[dict[str, Any], str, None] = None,
+    remote: Optional[str] = None,
+    remote_config: Optional[dict] = None,
+):
+    from dvc.config import Config
+
     from . import Repo
 
+    if config and not isinstance(config, dict):
+        config_dict = Config.load_file(config)
+    else:
+        config_dict = None
+
+    return Repo.open(
+        url,
+        rev=rev,
+        subrepos=True,
+        uninitialized=True,
+        config=config_dict,
+        remote=remote,
+        remote_config=remote_config,
+    )
+
+
+def _adapt_info(info: dict[str, Any]) -> dict[str, Any]:
+    dvc_info = info.get("dvc_info", {})
+    return {
+        "isout": dvc_info.get("isout", False),
+        "isdir": info["type"] == "directory",
+        "isexec": info.get("isexec", False),
+        "size": info.get("size"),
+        "md5": dvc_info.get("md5") or dvc_info.get("md5-dos2unix"),
+    }
+
 
 def ls(
     url: str,
@@ -13,9 +49,10 @@ def ls(
     rev: Optional[str] = None,
     recursive: Optional[bool] = None,
     dvc_only: bool = False,
-    config: Union[None, dict[str, Any], str] = None,
+    config: Union[dict[str, Any], str, None] = None,
     remote: Optional[str] = None,
     remote_config: Optional[dict] = None,
+    maxdepth: Optional[int] = None,
 ):
     """Methods for getting files and outputs for the repo.
 
@@ -41,60 +78,60 @@ def ls(
             "isexec": bool,
         }
     """
-    from dvc.config import Config
-
-    from . import 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:
+    with _open_repo(url, rev, config, remote, remote_config) as repo:
         path = path or ""
+        fs: DVCFileSystem = repo.dvcfs
+        fs_path = fs.from_os_path(path)
+        return _ls(fs, fs_path, recursive, dvc_only, maxdepth)
 
-        ret = _ls(repo, path, recursive, dvc_only)
 
-        ret_list = []
-        for path, info in ret.items():
-            info["path"] = path
-            ret_list.append(info)
-        ret_list.sort(key=lambda f: f["path"])
-        return ret_list
+def ls_tree(
+    url: str,
+    path: Optional[str] = None,
+    rev: Optional[str] = None,
+    dvc_only: bool = False,
+    config: Union[dict[str, Any], str, None] = None,
+    remote: Optional[str] = None,
+    remote_config: Optional[dict] = None,
+    maxdepth: Optional[int] = None,
+):
+    with _open_repo(url, rev, config, remote, remote_config) as repo:
+        path = path or ""
+        fs: DVCFileSystem = repo.dvcfs
+        fs_path = fs.from_os_path(path)
+        return _ls_tree(
+            fs, fs_path, maxdepth=maxdepth, dvc_only=dvc_only, dvcfiles=True
+        )
 
 
 def _ls(
-    repo: "Repo",
+    fs: "DVCFileSystem",
     path: str,
     recursive: Optional[bool] = None,
     dvc_only: bool = False,
+    maxdepth: Optional[int] = None,
 ):
-    fs: "DVCFileSystem" = repo.dvcfs
-    fs_path = fs.from_os_path(path)
-
-    fs_path = fs.info(fs_path)["name"]
+    fs_path = fs.info(path)["name"]
 
     infos = {}
-    if fs.isfile(fs_path):
-        infos[os.path.basename(path)] = fs.info(fs_path)
+
+    # ignore maxdepth only if recursive is not set
+    maxdepth = maxdepth if recursive else None
+    if maxdepth == 0 or fs.isfile(fs_path):
+        infos[os.path.basename(path) or os.curdir] = fs.info(fs_path)
     else:
         for root, dirs, files in fs.walk(
-            fs_path, dvcfiles=True, dvc_only=dvc_only, detail=True
+            fs_path,
+            dvcfiles=True,
+            dvc_only=dvc_only,
+            detail=True,
+            maxdepth=maxdepth,
         ):
-            if not recursive:
-                files.update(dirs)
-
             parts = fs.relparts(root, fs_path)
             if parts == (".",):
                 parts = ()
+            if not recursive or (maxdepth and len(parts) >= maxdepth - 1):
+                files.update(dirs)
 
             for name, entry in files.items():
                 infos[os.path.join(*parts, name)] = entry
@@ -102,14 +139,44 @@ def _ls(
             if not recursive:
                 break
 
-    ret = {}
-    for name, info in infos.items():
-        dvc_info = info.get("dvc_info", {})
-        ret[name] = {
-            "isout": dvc_info.get("isout", False),
-            "isdir": info["type"] == "directory",
-            "isexec": info.get("isexec", False),
-            "size": info.get("size"),
-        }
+    ret_list = []
+    for p, info in sorted(infos.items(), key=lambda x: x[0]):
+        _info = _adapt_info(info)
+        _info["path"] = p
+        ret_list.append(_info)
+    return ret_list
 
+
+def _ls_tree(fs, path, maxdepth=None, _info=None, **fs_kwargs):
+    info = _info or fs.info(path)
+    if _info is None:
+        # preserve the original path name
+        name = path
+        if not name:
+            name = os.curdir if fs.protocol == "local" else fs.root_marker
+        path = info["name"]
+    else:
+        name = path.rsplit(fs.sep, 1)[-1]
+
+    ret = {}
+    ls_info = _adapt_info(info)
+    ls_info["path"] = path
+
+    recurse = maxdepth is None or maxdepth > 0
+    if recurse and info["type"] == "directory":
+        try:
+            infos = fs.ls(path, detail=True, **fs_kwargs)
+        except FileNotFoundError:
+            # broken symlink?
+            infos = []
+
+        infos.sort(key=lambda f: f["name"])
+        maxdepth = maxdepth - 1 if maxdepth is not None else None
+        contents = {}
+        for info in infos:
+            d = _ls_tree(fs, info["name"], maxdepth=maxdepth, _info=info, **fs_kwargs)
+            contents.update(d)
+        ls_info["contents"] = contents
+
+    ret[name] = ls_info
     return ret

dvc/repo/ls_url.py

@@ -1,19 +1,30 @@
+from fsspec.implementations.local import LocalFileSystem as _LocalFileSystem
+
 from dvc.exceptions import URLMissingError
-from dvc.fs import parse_external_url
+from dvc.fs import LocalFileSystem, parse_external_url
 
 
-def ls_url(url, *, fs_config=None, recursive=False, config=None):
+def ls_url(url, *, fs_config=None, recursive=False, maxdepth=None, config=None):
     fs, fs_path = parse_external_url(url, fs_config=fs_config, config=config)
     try:
         info = fs.info(fs_path)
     except FileNotFoundError as exc:
         raise URLMissingError(url) from exc
-    if info["type"] != "directory":
+    if maxdepth == 0 or info["type"] != "directory":
         return [{"path": info["name"], "isdir": False}]
 
+    if isinstance(fs, LocalFileSystem):
+        # dvc's LocalFileSystem does not support maxdepth yet
+        walk = _LocalFileSystem().walk
+    else:
+        walk = fs.walk
+
     ret = []
-    for _, dirs, files in fs.walk(fs_path, detail=True):
-        if not recursive:
+    for root, dirs, files in walk(fs_path, detail=True, maxdepth=maxdepth):
+        parts = fs.relparts(root, fs_path)
+        if parts == (".",):
+            parts = ()
+        if not recursive or (maxdepth and len(parts) >= maxdepth - 1):
             files.update(dirs)
 
         for info in files.values():

dvc/repo/metrics/show.py

@@ -1,3 +1,4 @@
+import logging
 import os
 from collections.abc import Iterable, Iterator
 from itertools import chain
@@ -8,7 +9,7 @@ from scmrepo.exceptions import SCMError
 
 from dvc.log import logger
 from dvc.scm import NoSCMError
-from dvc.utils import as_posix, expand_paths
+from dvc.utils import as_posix
 from dvc.utils.collections import ensure_list
 from dvc.utils.serialize import load_path
 
@@ -103,7 +104,7 @@ def _collect_metrics(
     paths = (fs.from_os_path(metric) for metric in metrics)
     # make paths absolute for DVCFileSystem
     repo_paths = (f"{fs.root_marker}{path}" for path in paths)
-    return ldistinct(expand_paths(fs, repo_paths))
+    return ldistinct(try_expand_paths(fs, repo_paths))
 
 
 class FileResult(TypedDict, total=False):
@@ -116,6 +117,22 @@ class Result(TypedDict, total=False):
     error: Exception
 
 
+def try_expand_paths(fs: "FileSystem", paths: Iterable[str]) -> Iterator[str]:
+    for path in paths:
+        try:
+            if fs.isdir(path):
+                yield from fs.find(path)
+                continue
+        except Exception as e:  # noqa: BLE001
+            logger.debug(
+                "failed to expand %r: %s",
+                path,
+                e,
+                exc_info=logger.isEnabledFor(logging.TRACE),  # type: ignore[attr-defined]
+            )
+        yield path
+
+
 def to_relpath(fs: "FileSystem", root_dir: str, d: Result) -> Result:
     relpath = fs.relpath
     cwd = fs.getcwd()
@@ -207,7 +224,7 @@ def show(
                 on_error=on_error,
             )
             res[rev] = Result(data=result)
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             if on_error == "raise":
                 raise
 

dvc/repo/open_repo.py

@@ -99,7 +99,7 @@ def clean_repos():
 
 def _get_remote_config(url):
     try:
-        repo = Repo(url)
+        repo = Repo(url, uninitialized=True)
     except NotDvcRepoError:
         return {}
 
@@ -109,10 +109,16 @@ def _get_remote_config(url):
             # Fill the empty upstream entry with a new remote pointing to the
             # original repo's cache location.
             name = "auto-generated-upstream"
-            return {
-                "core": {"remote": name},
-                "remote": {name: {"url": repo.cache.local_cache_dir}},
-            }
+            try:
+                local_cache_dir = repo.cache.local_cache_dir
+            except AttributeError:
+                # if the `.dvc` dir is missing, we get an AttributeError
+                return {}
+            else:
+                return {
+                    "core": {"remote": name},
+                    "remote": {name: {"url": local_cache_dir}},
+                }
 
         # Use original remote to make sure that we are using correct url,
         # credential paths, etc if they are relative to the config location.
@@ -146,7 +152,7 @@ def _cached_clone(url, rev):
 
 
 @wrap_with(threading.Lock())
-def _clone_default_branch(url, rev):  # noqa: PLR0912
+def _clone_default_branch(url, rev):
     """Get or create a clean clone of the url.
 
     The cloned is reactualized with git pull unless rev is a known sha.

dvc/repo/params/show.py

@@ -6,9 +6,9 @@ from typing import TYPE_CHECKING, Any, Optional, Union
 
 from dvc.dependency.param import ParamsDependency, read_param_file
 from dvc.log import logger
-from dvc.repo.metrics.show import FileResult, Result
+from dvc.repo.metrics.show import FileResult, Result, try_expand_paths
 from dvc.stage import PipelineStage
-from dvc.utils import as_posix, expand_paths
+from dvc.utils import as_posix
 from dvc.utils.collections import ensure_list
 
 if TYPE_CHECKING:
@@ -77,7 +77,7 @@ def _collect_params(
         path = fs.from_os_path(param)
         # make paths absolute for DVCFileSystem
         repo_path = f"{fs.root_marker}{path}"
-        ret.update(dict.fromkeys(expand_paths(fs, [repo_path]), _params))
+        ret.update(dict.fromkeys(try_expand_paths(fs, [repo_path]), _params))
     return ret
 
 
@@ -191,7 +191,7 @@ def show(
                 on_error=on_error,
             )
             res[rev] = Result(data=params)
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             if on_error == "raise":
                 raise
             logger.warning("failed to load params in revision %r, %s", rev, str(exc))

dvc/repo/push.py

@@ -1,10 +1,14 @@
 from contextlib import suppress
 
 from dvc.exceptions import InvalidArgumentError, UploadError
+from dvc.log import logger
+from dvc.stage.cache import RunCacheNotSupported
 from dvc.ui import ui
 
 from . import locked
 
+logger = logger.getChild(__name__)
+
 
 def _rebuild(idx, path, fs, cb):
     from dvc_data.index import DataIndex, DataIndexEntry, Meta
@@ -95,8 +99,11 @@ def push(  # noqa: PLR0913
                 "Multiple rev push is unsupported for cloud versioned remotes"
             )
 
-    used_run_cache = self.stage_cache.push(remote) if run_cache else []
-    transferred_count += len(used_run_cache)
+    try:
+        used_run_cache = self.stage_cache.push(remote) if run_cache else []
+        transferred_count += len(used_run_cache)
+    except RunCacheNotSupported as e:
+        logger.debug("failed to push run cache: %s", e)
 
     if isinstance(targets, str):
         targets = [targets]

dvc/repo/reproduce.py

@@ -29,7 +29,7 @@ def collect_stages(
     recursive: bool = False,
     glob: bool = False,
 ) -> list["Stage"]:
-    stages: list["Stage"] = []
+    stages: list[Stage] = []
     for target in targets:
         stages.extend(repo.stage.collect(target, recursive=recursive, glob=glob))
     return ldistinct(stages)
@@ -166,10 +166,10 @@ def _reproduce(
 ) -> list["Stage"]:
     assert on_error in ("fail", "keep-going", "ignore")
 
-    result: list["Stage"] = []
-    failed: list["Stage"] = []
-    to_skip: dict["Stage", "Stage"] = {}
-    ret: Optional["Stage"] = None
+    result: list[Stage] = []
+    failed: list[Stage] = []
+    to_skip: dict[Stage, Stage] = {}
+    ret: Optional[Stage] = None
 
     force_state = dict.fromkeys(stages, force)
 
@@ -228,7 +228,7 @@ def reproduce(
     if not kwargs.get("interactive", False):
         kwargs["interactive"] = self.config["core"].get("interactive", False)
 
-    stages: list["Stage"] = []
+    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)

dvc/repo/scm_context.py

@@ -21,7 +21,7 @@ class SCMContext:
     def __init__(self, scm: "Base", config: Optional[dict[str, Any]] = None) -> None:
         from funcy import get_in
 
-        self.scm: "Base" = scm
+        self.scm: Base = scm
         self.autostage: bool = get_in(
             config or {}, ["core", "autostage"], default=False
         )

dvc/repo/stage.py

@@ -108,7 +108,7 @@ def locked(f):
 
 class StageLoad:
     def __init__(self, repo: "Repo") -> None:
-        self.repo: "Repo" = repo
+        self.repo: Repo = repo
 
     @property
     def fs(self):

dvc/repo/update.py

@@ -34,7 +34,7 @@ def update(  # noqa: C901
         raise InvalidArgumentError("--remote can't be used without --to-remote")
 
     import_stages = set()
-    other_stage_infos: list["StageInfo"] = []
+    other_stage_infos: list[StageInfo] = []
 
     for stage_info in self.index.collect_targets(targets, recursive=recursive):
         if stage_info.stage.is_import:

dvc/repo/worktree.py

@@ -166,7 +166,7 @@ def update_worktree_stages(repo: "Repo", stage_infos: Iterable["StageInfo"]):
 
     view = IndexView(repo.index, stage_infos, outs_filter=outs_filter)
     local_index = view.data["repo"]
-    remote_indexes: dict[str, tuple["Remote", "DataIndex"]] = {}
+    remote_indexes: dict[str, tuple[Remote, DataIndex]] = {}
     for stage in view.stages:
         for out in stage.outs:
             _update_worktree_out(repo, out, local_index, remote_indexes)

dvc/stage/__init__.py

@@ -83,7 +83,7 @@ class RawData:
 def create_stage(cls: type[_T], repo, path, **kwargs) -> _T:
     from dvc.dvcfile import check_dvcfile_path
 
-    wdir = os.path.abspath(kwargs.get("wdir", None) or os.curdir)
+    wdir = os.path.abspath(kwargs.get("wdir") or os.curdir)
     path = os.path.abspath(path)
 
     check_dvcfile_path(repo, path)
@@ -290,7 +290,7 @@ class Stage(params.StageParams):
         )
 
     def short_description(self) -> Optional["str"]:
-        desc: Optional["str"] = None
+        desc: Optional[str] = None
         if self.desc:
             with suppress(ValueError):
                 # try to use first non-empty line as a description
@@ -339,7 +339,7 @@ class Stage(params.StageParams):
         for out in self.outs:
             status = out.status()
             if status:
-                if allow_missing and status[str(out)] == "not in cache":
+                if allow_missing and status[str(out)] in ["not in cache", "deleted"]:
                     continue
                 logger.debug(
                     "Output '%s' of %s changed because it is '%s'.",
@@ -418,17 +418,28 @@ class Stage(params.StageParams):
 
     @rwlocked(read=["deps"], write=["outs"])
     def reproduce(self, interactive=False, **kwargs) -> Optional["Stage"]:
-        if not (
-            kwargs.get("force", False)
-            or self.changed(
-                kwargs.get("allow_missing", False), kwargs.pop("upstream", None)
-            )
-        ):
+        force = kwargs.get("force", False)
+        allow_missing = kwargs.get("allow_missing", False)
+        pull = kwargs.get("pull", False)
+        upstream = kwargs.pop("upstream", None)
+        if force:
+            pass
+        # Skip stages with missing data if otherwise unchanged
+        elif not self.changed(allow_missing, upstream):
             if not isinstance(self, PipelineStage) and self.is_data_source:
                 logger.info("'%s' didn't change, skipping", self.addressing)
             else:
                 logger.info("Stage '%s' didn't change, skipping", self.addressing)
             return None
+        # Pull stages with missing data if otherwise unchanged
+        elif not self.changed(True, upstream) and pull:
+            try:
+                logger.info("Pulling data for %s", self)
+                self.repo.pull(self.addressing, jobs=kwargs.get("jobs"))
+                self.checkout()
+                return None
+            except CheckoutError:
+                logger.info("Unable to pull data for %s", self)
 
         msg = f"Going to reproduce {self}. Are you sure you want to continue?"
         if interactive and not prompt.confirm(msg):
@@ -597,16 +608,10 @@ class Stage(params.StageParams):
         if (self.cmd or self.is_import) and not self.frozen and not dry:
             self.remove_outs(ignore_remove=False, force=False)
 
-        if (
-            self.is_import and (not self.frozen or kwargs.get("pull"))
-        ) or self.is_partial_import:
-            self._sync_import(dry, force, kwargs.get("jobs", None), no_download)
+        if (self.is_import and not self.frozen) or self.is_partial_import:
+            self._sync_import(dry, force, kwargs.get("jobs"), no_download)
         elif not self.frozen and self.cmd:
             self._run_stage(dry, force, **kwargs)
-        elif kwargs.get("pull"):
-            logger.info("Pulling data for %s", self)
-            self.repo.pull(self.addressing, jobs=kwargs.get("jobs", None))
-            self.checkout()
         elif not dry:
             args = ("outputs", "frozen ") if self.frozen else ("data sources", "")
             logger.info("Verifying %s in %s%s", *args, self)
@@ -653,7 +658,7 @@ class Stage(params.StageParams):
     def checkout(
         self, allow_missing: bool = False, **kwargs
     ) -> dict[str, list["StrPath"]]:
-        stats: dict[str, list["StrPath"]] = defaultdict(list)
+        stats: dict[str, list[StrPath]] = defaultdict(list)
         if self.is_partial_import:
             return stats