import/get: escape non-pattern LFS path parts (#10416)

* [pre-commit.ci] pre-commit autoupdate

updates:
- [github.com/astral-sh/ruff-pre-commit: v0.4.1 → v0.4.3](https://github.com/astral-sh/ruff-pre-commit/compare/v0.4.1...v0.4.3)

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

.flake8

@@ -1,20 +0,0 @@
-[flake8]
-max_line_length = 88
-extend-ignore=
-    # Whitespace before ':'
-    E203
-    # Line break occurred before a binary operator
-    W503
-    # unindexed parameters in the str.format, see:
-    # https://pypi.org/project/flake8-string-format/
-    P1
-max-complexity = 10
-show_source = true
-count = true
-min_python_version = 3.8.0
-per-file-ignores =
-    dvc/testing/benchmarks/conftest.py:F401
-    dvc/testing/conftest.py:F401
-    dvc/testing/plugin.py:F401
-    tests/conftest.py:F401
-    tests/func/experiments/executor/test_ssh.py:F401

.github/mergify.yml

@@ -1,10 +0,0 @@
-pull_request_rules:
-  - name: backport patches to 2.x
-    conditions:
-      - base=main
-      - label=backport
-    actions:
-      backport:
-        branches:
-          - "2.x"
-        title: ({{ destination_branch }}) {{ title }}

.github/workflows/benchmarks.yaml

@@ -10,7 +10,7 @@ jobs:
     name: run benchmarks
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/setup-python@v4
+      - uses: actions/setup-python@v5
         with:
             python-version: "3.11"
       - uses: iterative/dvc-bench@main

.github/workflows/build.yaml

@@ -23,9 +23,9 @@ jobs:
       with:
         fetch-depth: 0
 
-    - uses: actions/setup-python@v4
+    - uses: actions/setup-python@v5
       with:
-        python-version: 3.8
+        python-version: 3.12
 
     - run: python -m pip install --upgrade pip setuptools_scm build twine
 
@@ -39,7 +39,7 @@ jobs:
         SETUPTOOLS_SCM_PRETEND_VERSION_FOR_DVC: ${{ env.version }}
     - run: twine check --strict dist/*
 
-    - uses: actions/upload-artifact@v3
+    - uses: actions/upload-artifact@v4
       with:
         name: Packages
         path: dist/
@@ -58,7 +58,7 @@ jobs:
       id-token: write
 
     steps:
-    - uses: actions/download-artifact@v3
+    - uses: actions/download-artifact@v4
       with:
         name: Packages
         path: dist
@@ -83,7 +83,7 @@ jobs:
       id-token: write
 
     steps:
-    - uses: actions/download-artifact@v3
+    - uses: actions/download-artifact@v4
       with:
         name: Packages
         path: dist

.github/workflows/codeql.yml

@@ -2,7 +2,7 @@ name: "CodeQL"
 
 on:
   push:
-    branches: ["main", "2.x"]
+    branches: ["main"]
   pull_request:
     branches: ["main"]
   schedule:
@@ -31,9 +31,10 @@ jobs:
       uses: actions/checkout@v4
 
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@v2
+      uses: github/codeql-action/init@v3
       with:
         languages: ${{ matrix.language }}
+        queries: security-extended
 
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@v2
+      uses: github/codeql-action/analyze@v3

.github/workflows/plugin_tests.yaml

@@ -34,7 +34,7 @@ jobs:
         path: ${{ matrix.plugin }}
 
     - name: Set up Python
-      uses: actions/setup-python@v4
+      uses: actions/setup-python@v5
       with:
         python-version: ${{ matrix.pyv }}
         cache: pip

.github/workflows/tests.yaml

@@ -2,7 +2,7 @@ name: Tests
 
 on:
   push:
-    branches: [main, 2.x]
+    branches: [main]
   pull_request:
   schedule:
     - cron: '5 1 * * *'  # every day at 01:05
@@ -23,14 +23,14 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-20.04, windows-latest, macos-latest]
-        pyv: ["3.8", "3.9", "3.10", "3.11"]
+        pyv: ["3.9", "3.10", "3.11", "3.12"]
 
     steps:
     - uses: actions/checkout@v4
       with:
         fetch-depth: 0
     - name: Set up Python
-      uses: actions/setup-python@v4
+      uses: actions/setup-python@v5
       with:
         python-version: ${{ matrix.pyv }}
         cache: pip
@@ -39,7 +39,7 @@ jobs:
       run: |
         pip install --upgrade pip wheel
         pip install -e ".[dev]"
-    - uses: pre-commit/[email protected].0
+    - uses: pre-commit/[email protected].1
     - name: run tests
       timeout-minutes: 40
       env:
@@ -66,7 +66,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: Slack Notification
-      uses: rtCamp/action-slack-notify@v2.2.1
+      uses: rtCamp/action-slack-notify@v2.3.0
       env:
         SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}
         SLACK_COLOR: ${{ job.status }}

.pre-commit-config.yaml

@@ -2,11 +2,11 @@ default_language_version:
   python: python3
 
 ci:
-  skip: [mypy, pylint, dvc-pre-commit]
+  skip: [mypy, dvc-pre-commit]
 
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.4.0
+    rev: v4.6.0
     hooks:
       - id: check-added-large-files
       - id: check-case-conflict
@@ -17,44 +17,21 @@ repos:
         args: ['--assume-in-merge']
       - id: check-toml
       - id: check-yaml
-      - id: debug-statements
-        exclude: "dvc/_debug.py"
       - id: end-of-file-fixer
       - id: mixed-line-ending
       - id: sort-simple-yaml
       - id: trailing-whitespace
-  - repo: https://github.com/psf/black
-    rev: 23.9.1
-    hooks:
-      - id: black
-  - repo: https://github.com/PyCQA/isort
-    rev: 5.12.0
-    hooks:
-      - id: isort
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.0.291'
+    rev: 'v0.4.4'
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]
+      - id: ruff-format
   - repo: https://github.com/codespell-project/codespell
-    rev: v2.2.5
+    rev: v2.2.6
     hooks:
       - id: codespell
         additional_dependencies: ["tomli"]
-  - repo: https://github.com/PyCQA/flake8
-    rev: 6.1.0
-    hooks:
-      - id: flake8
-        additional_dependencies:
-          - flake8-string-format
-          - flake8-typing-imports
-  - repo: https://github.com/PyCQA/bandit
-    rev: '1.7.5'
-    hooks:
-    - id: bandit
-      args: ["-c", "pyproject.toml", "--quiet", "--format=custom"]
-      files: ^dvc/
-      additional_dependencies: [".[toml]"]
   - repo: local
     hooks:
     - id: mypy
@@ -64,13 +41,6 @@ repos:
       language: system
       types: [python]
       require_serial: true
-    - id: pylint
-      name: pylint
-      entry: pylint
-      files: ^dvc/
-      language: system
-      types: [python]
-      require_serial: true
   - hooks:
       - args:
           - git-hook

CITATION.cff

@@ -0,0 +1,25 @@
+cff-version: 1.2.0
+title: 'DVC: Data Version Control - Git for Data & Models"'
+message: >-
+  If you use this software as part of a publication and wish
+  to cite it, please use the metadata from this file.
+type: software
+authors:
+  - name: The DVC team and contributors
+    website: 'https://github.com/iterative'
+repository-code: 'https://github.com/iterative/dvc'
+url: 'https://dvc.org'
+abstract: >-
+  DVC is a tool for data science that takes advantage of existing software engineering toolset.
+  It helps machine learning teams manage large datasets, make projects reproducible, and collaborate better.
+keywords:
+  - ai
+  - collaboration
+  - data-science
+  - data-version-control
+  - developer-tools
+  - git
+  - machine-learning
+  - python
+  - reproducibility
+license: Apache-2.0

README.rst

@@ -90,8 +90,8 @@ Their results can be filtered and compared based on hyperparameters and metrics,
 
 .. _`VS Code Extension`:
 
-Visual Studio Code Extension
-============================
+VS Code Extension
+=================
 
 |VS Code|
 

dvc/__init__.py

@@ -3,6 +3,7 @@ DVC
 ----
 Make your data science projects reproducible and shareable.
 """
+
 import dvc.logger
 from dvc.build import PKG  # noqa: F401
 from dvc.version import __version__, version_tuple  # noqa: F401

dvc/__main__.py

@@ -1,4 +1,5 @@
 """Main entry point for DVC command line tool."""
+
 import sys
 
 from dvc.cli import main

dvc/__pyinstaller/hook-celery.py

@@ -1,14 +1,11 @@
 # ruff: noqa: N999
-# pylint: disable=import-error
+
 from PyInstaller.utils.hooks import collect_submodules, is_module_or_submodule
 
 # Celery dynamically imports most celery internals at runtime
 # pyinstaller hook must expose all modules loaded by
 # kombu.utils.imports:symbol_by_name()
-_EXCLUDES = (
-    "celery.bin",
-    "celery.contrib",
-)
+_EXCLUDES = ("celery.bin", "celery.contrib")
 hiddenimports = collect_submodules(
     "celery",
     filter=lambda name: not any(

dvc/__pyinstaller/hook-dvc.py

@@ -1,5 +1,5 @@
 # ruff: noqa: N999
-from PyInstaller.utils.hooks import copy_metadata  # pylint:disable=import-error
+from PyInstaller.utils.hooks import copy_metadata
 
 # needed for `dvc doctor` to show dep versions
 datas = copy_metadata("adlfs", recursive=True)

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

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

dvc/__pyinstaller/hook-dvc_task.py

@@ -1,5 +1,5 @@
 # ruff: noqa: N999
-# pylint: disable=import-error
+
 from PyInstaller.utils.hooks import collect_submodules
 
 hiddenimports = collect_submodules("dvc_task")

dvc/__pyinstaller/hook-pydrive2.py

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

dvc/_debug.py

@@ -14,7 +14,7 @@ def viztracer_profile(
     log_async: bool = False,
 ):
     try:
-        import viztracer  # pylint: disable=import-error
+        import viztracer
     except ImportError:
         print("Failed to run profiler, viztracer is not installed")  # noqa: T201
         yield
@@ -23,10 +23,11 @@ def viztracer_profile(
     tracer = viztracer.VizTracer(max_stack_depth=depth, log_async=log_async)
 
     tracer.start()
-    yield
-    tracer.stop()
-
-    tracer.save(path() if callable(path) else path)
+    try:
+        yield
+    finally:
+        tracer.stop()
+        tracer.save(path() if callable(path) else path)
 
 
 @contextmanager
@@ -36,7 +37,7 @@ def yappi_profile(
     separate_threads: Optional[bool] = False,
 ):
     try:
-        import yappi  # pylint: disable=import-error
+        import yappi
     except ImportError:
         print("Failed to run profiler, yappi is not installed")  # noqa: T201
         yield
@@ -45,38 +46,40 @@ def yappi_profile(
     yappi.set_clock_type("wall" if wall_clock else "cpu")
 
     yappi.start()
-    yield
-    yappi.stop()
+    try:
+        yield
+    finally:
+        yappi.stop()
 
-    threads = yappi.get_thread_stats()
-    stats = {}
-    if separate_threads:
-        for thread in threads:
-            ctx_id = thread.id
-            stats[ctx_id] = yappi.get_func_stats(ctx_id=ctx_id)
-    else:
-        stats[None] = yappi.get_func_stats()
-
-    fpath = path() if callable(path) else path
-    for ctx_id, st in stats.items():
-        if fpath:
-            out = f"{fpath}-{ctx_id}" if ctx_id is not None else fpath
-            st.save(out, type="callgrind")
+        threads = yappi.get_thread_stats()
+        stats = {}
+        if separate_threads:
+            for thread in threads:
+                ctx_id = thread.id
+                stats[ctx_id] = yappi.get_func_stats(ctx_id=ctx_id)
         else:
-            if ctx_id is not None:
-                print(f"\nThread {ctx_id}")  # noqa: T201
-            st.print_all()  # pylint:disable=no-member
-            if ctx_id is None:
-                threads.print_all()  # pylint:disable=no-member
+            stats[None] = yappi.get_func_stats()
 
-    yappi.clear_stats()
+        fpath = path() if callable(path) else path
+        for ctx_id, st in stats.items():
+            if fpath:
+                out = f"{fpath}-{ctx_id}" if ctx_id is not None else fpath
+                st.save(out, type="callgrind")
+            else:
+                if ctx_id is not None:
+                    print(f"\nThread {ctx_id}")  # noqa: T201
+                st.print_all()
+                if ctx_id is None:
+                    threads.print_all()
+
+        yappi.clear_stats()
 
 
 @contextmanager
 def instrument(html_output=False):
     """Run a statistical profiler"""
     try:
-        from pyinstrument import Profiler  # pylint: disable=import-error
+        from pyinstrument import Profiler
     except ImportError:
         print("Failed to run profiler, pyinstrument is not installed")  # noqa: T201
         yield
@@ -85,13 +88,15 @@ def instrument(html_output=False):
     profiler = Profiler()
 
     profiler.start()
-    yield
-    profiler.stop()
+    try:
+        yield
+    finally:
+        profiler.stop()
 
-    if html_output:
-        profiler.open_in_browser()
-        return
-    print(profiler.output_text(unicode=True, color=True))  # noqa: T201
+        if html_output:
+            profiler.open_in_browser()
+        else:
+            print(profiler.output_text(unicode=True, color=True))  # noqa: T201
 
 
 @contextmanager
@@ -102,22 +107,23 @@ def profile(dump_path: Optional[str] = None):
     prof = cProfile.Profile()
     prof.enable()
 
-    yield
-
-    prof.disable()
-    if not dump_path:
-        prof.print_stats(sort="cumtime")
-        return
-    prof.dump_stats(dump_path)
+    try:
+        yield
+    finally:
+        prof.disable()
+        if dump_path:
+            prof.dump_stats(dump_path)
+        else:
+            prof.print_stats(sort="cumtime")
 
 
 @contextmanager
 def debug():
     try:
         yield
-    except Exception:  # pylint: disable=broad-except
+    except Exception:
         try:
-            import ipdb as pdb  # noqa: T100, pylint: disable=import-error
+            import ipdb as pdb  # noqa: T100
         except ImportError:
             import pdb  # type: ignore[no-redef]  # noqa: T100
         pdb.post_mortem()
@@ -145,7 +151,8 @@ def show_stack():
         signal.signal(signal.SIGQUIT, _sigshow)
 
     try:
-        signal.signal(signal.SIGINFO, _sigshow)  # only available on macOS
+        # only available on macOS
+        signal.signal(signal.SIGINFO, _sigshow)  # type: ignore[attr-defined]
     except AttributeError:
         pass
     yield
@@ -155,7 +162,7 @@ def _get_path_func(tool: str, ext: str):
     fmt = f"{tool}.dvc-{{now:%Y%m%d}}_{{now:%H%M%S}}.{ext}"
 
     def func(now: Optional["datetime"] = None) -> str:
-        return fmt.format(now=now or datetime.now())
+        return fmt.format(now=now or datetime.now())  # noqa: DTZ005
 
     return func
 

dvc/analytics.py

@@ -1,10 +1,11 @@
 import json
-import logging
 import os
 
+from dvc.log import logger
+
 from .env import DVC_ANALYTICS_ENDPOINT, DVC_NO_ANALYTICS
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 def collect_and_send_report(args=None, return_code=None):
@@ -35,9 +36,7 @@ def collect_and_send_report(args=None, return_code=None):
     with tempfile.NamedTemporaryFile(delete=False, mode="w") as fobj:
         json.dump(report, fobj)
 
-    logger.trace(  # type: ignore[attr-defined]
-        "Saving analytics report to %s", fobj.name
-    )
+    logger.trace("Saving analytics report to %s", fobj.name)
     daemon(["analytics", fobj.name])
 
 
@@ -78,15 +77,15 @@ def send(path):
     report.update(_runtime_info())
 
     logger.debug("uploading report to %s", url)
-    logger.trace("Sending %s to %s", report, url)  # type: ignore[attr-defined]
+    logger.trace("Sending %s to %s", report, url)
 
     try:
         requests.post(url, json=report, headers=headers, timeout=5)
     except requests.exceptions.RequestException as e:
-        logger.trace("", exc_info=True)  # type: ignore[attr-defined]
+        logger.trace("", exc_info=True)
         logger.debug("failed to send analytics report %s", str(e))
 
-    logger.trace("removing report %s", path)  # type: ignore[attr-defined]
+    logger.trace("removing report %s", path)
     os.remove(path)
 
 

dvc/annotations.py

@@ -1,5 +1,5 @@
 from dataclasses import asdict, dataclass, field, fields
-from typing import Any, ClassVar, Dict, List, Optional
+from typing import Any, ClassVar, Optional
 
 from funcy import compact
 from voluptuous import Required
@@ -13,11 +13,11 @@ class Annotation:
     PARAM_META: ClassVar[str] = "meta"
 
     desc: Optional[str] = None
-    type: Optional[str] = None  # noqa: A003
-    labels: List[str] = field(default_factory=list)
-    meta: Dict[str, Any] = field(default_factory=dict)
+    type: Optional[str] = None
+    labels: list[str] = field(default_factory=list)
+    meta: dict[str, Any] = field(default_factory=dict)
 
-    def to_dict(self) -> Dict[str, str]:
+    def to_dict(self) -> dict[str, str]:
         return compact(asdict(self))
 
 
@@ -31,11 +31,11 @@ class Artifact:
 
     path: str
     desc: Optional[str] = None
-    type: Optional[str] = None  # noqa: A003
-    labels: List[str] = field(default_factory=list)
-    meta: Dict[str, Any] = field(default_factory=dict)
+    type: Optional[str] = None
+    labels: list[str] = field(default_factory=list)
+    meta: dict[str, Any] = field(default_factory=dict)
 
-    def to_dict(self) -> Dict[str, str]:
+    def to_dict(self) -> dict[str, str]:
         return compact(asdict(self))
 
 
@@ -46,7 +46,7 @@ ANNOTATION_SCHEMA = {
     Annotation.PARAM_LABELS: [str],
     Annotation.PARAM_META: object,
 }
-ARTIFACT_SCHEMA = {
+ARTIFACT_SCHEMA: dict[Any, Any] = {
     Required(Artifact.PARAM_PATH): str,
     **ANNOTATION_SCHEMA,  # type: ignore[arg-type]
 }

dvc/api/__init__.py

@@ -1,23 +1,24 @@
 from dvc.fs.dvc import _DVCFileSystem as DVCFileSystem
 
 from .artifacts import artifacts_show
-from .data import open  # pylint: disable=redefined-builtin
-from .data import get_url, read
+from .data import get_url, open, read
+from .dataset import get as get_dataset
 from .experiments import exp_save, exp_show
 from .scm import all_branches, all_commits, all_tags
 from .show import metrics_show, params_show
 
 __all__ = [
+    "DVCFileSystem",
     "all_branches",
     "all_commits",
     "all_tags",
     "artifacts_show",
     "exp_save",
     "exp_show",
+    "get_dataset",
     "get_url",
+    "metrics_show",
     "open",
     "params_show",
-    "metrics_show",
     "read",
-    "DVCFileSystem",
 ]

dvc/api/artifacts.py

@@ -1,4 +1,5 @@
-from typing import Any, Dict, Optional
+import os
+from typing import Any, Optional
 
 from dvc.repo import Repo
 
@@ -8,7 +9,7 @@ def artifacts_show(
     version: Optional[str] = None,
     stage: Optional[str] = None,
     repo: Optional[str] = None,
-) -> Dict[str, str]:
+) -> dict[str, str]:
     """
     Return path and Git revision for an artifact in a DVC project.
 
@@ -36,12 +37,22 @@ def artifacts_show(
     if version and stage:
         raise ValueError("Artifact version and stage are mutually exclusive.")
 
-    repo_kwargs: Dict[str, Any] = {
+    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

@@ -1,6 +1,6 @@
 from contextlib import _GeneratorContextManager as GCM
 from contextlib import contextmanager
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 
 from funcy import reraise
 
@@ -25,7 +25,14 @@ def _wrap_exceptions(repo, url):
         raise PathMissingError(exc.path, url) from exc
 
 
-def get_url(path, repo=None, rev=None, remote=None):
+def get_url(
+    path: str,
+    repo: Optional[str] = None,
+    rev: Optional[str] = None,
+    remote: Optional[str] = None,
+    config: Optional[dict[str, Any]] = None,
+    remote_config: Optional[dict[str, Any]] = None,
+):
     """
     Returns the URL to the storage location of a data file or directory tracked
     in a DVC repo. For Git repos, HEAD is used unless a rev argument is
@@ -35,15 +42,44 @@ def get_url(path, repo=None, rev=None, remote=None):
 
     NOTE: This function does not check for the actual existence of the file or
     directory in the remote storage.
+
+    Args:
+        path (str): location and file name of the target, relative to the root
+            of `repo`.
+        repo (str, optional): location of the DVC project or Git Repo.
+            Defaults to the current DVC project (found by walking up from the
+            current working directory tree).
+            It can be a URL or a file system path.
+            Both HTTP and SSH protocols are supported for online Git repos
+            (e.g. [user@]server:project.git).
+        rev (str, optional): Any `Git revision`_ such as a branch or tag name,
+            a commit hash or a dvc experiment name.
+            Defaults to HEAD.
+            If `repo` is not a Git repo, this option is ignored.
+        remote (str, optional): Name of the `DVC remote`_ used to form the
+            returned URL string.
+            Defaults to the `default remote`_ of `repo`.
+            For local projects, the cache is tried before the default remote.
+        config(dict, optional): config to be passed to the DVC repository.
+            Defaults to None.
+        remote_config(dict, optional): remote config to be passed to the DVC
+            repository.
+            Defaults to None.
+
+    Returns:
+        str: URL to the file or directory.
     """
     from dvc.config import NoRemoteError
     from dvc_data.index import StorageKeyError
 
-    repo_kwargs: Dict[str, Any] = {}
-    if remote:
-        repo_kwargs["config"] = {"core": {"remote": remote}}
     with Repo.open(
-        repo, rev=rev, subrepos=True, uninitialized=True, **repo_kwargs
+        repo,
+        rev=rev,
+        subrepos=True,
+        uninitialized=True,
+        remote=remote,
+        config=config,
+        remote_config=remote_config,
     ) as _repo:
         index, entry = _repo.get_data_index_entry(path)
         with reraise(
@@ -55,27 +91,23 @@ def get_url(path, repo=None, rev=None, remote=None):
 
 
 class _OpenContextManager(GCM):
-    def __init__(self, func, args, kwds):  # pylint: disable=super-init-not-called
+    def __init__(self, func, args, kwds):
         self.gen = func(*args, **kwds)
-        self.func, self.args, self.kwds = (  # type: ignore[assignment]
-            func,
-            args,
-            kwds,
-        )
+        self.func, self.args, self.kwds = (func, args, kwds)  # type: ignore[assignment]
 
     def __getattr__(self, name):
         raise AttributeError("dvc.api.open() should be used in a with statement.")
 
 
-def open(  # noqa: A001, pylint: disable=redefined-builtin
+def open(  # noqa: A001
     path: str,
     repo: Optional[str] = None,
     rev: Optional[str] = None,
     remote: Optional[str] = None,
     mode: str = "r",
     encoding: Optional[str] = None,
-    config: Optional[Dict[str, Any]] = None,
-    remote_config: Optional[Dict[str, Any]] = None,
+    config: Optional[dict[str, Any]] = None,
+    remote_config: Optional[dict[str, Any]] = None,
 ):
     """
     Opens a file tracked in a DVC project.
@@ -233,7 +265,7 @@ def _open(
     config=None,
     remote_config=None,
 ):
-    repo_kwargs: Dict[str, Any] = {
+    repo_kwargs: dict[str, Any] = {
         "subrepos": True,
         "uninitialized": True,
         "remote": remote,
@@ -262,11 +294,7 @@ def _open(
                 fs_path = fs.from_os_path(path)
 
             try:
-                with fs.open(
-                    fs_path,
-                    mode=mode,
-                    encoding=encoding,
-                ) as fobj:
+                with fs.open(fs_path, mode=mode, encoding=encoding) as fobj:
                     yield fobj
             except FileNotFoundError as exc:
                 raise FileMissingError(path) from exc

dvc/api/dataset.py

@@ -0,0 +1,66 @@
+from typing import Literal, TypedDict, Union
+
+
+class DVCXDataset(TypedDict):
+    type: Literal["dvcx"]
+    name: str
+    version: int
+
+
+class DVCDataset(TypedDict):
+    type: Literal["dvc"]
+    url: str
+    path: str
+    sha: str
+
+
+class URLDataset(TypedDict):
+    type: Literal["url"]
+    files: list[str]
+    path: str
+
+
+def get(name: str) -> Union[DVCXDataset, DVCDataset, URLDataset]:
+    from difflib import get_close_matches
+
+    from dvc.fs import get_cloud_fs
+    from dvc.repo import Repo, datasets
+
+    repo = Repo()
+    try:
+        dataset = repo.datasets[name]
+    except datasets.DatasetNotFoundError as e:
+        add_note = getattr(e, "add_note", lambda _: None)
+        if matches := get_close_matches(name, repo.datasets):
+            add_note(f"Did you mean: {matches[0]!r}?")
+        raise
+
+    if dataset._invalidated:
+        raise ValueError(f"dataset not in sync. Sync with 'dvc ds update {name}'.")
+    if not dataset.lock:
+        raise ValueError("missing lock information")
+    if dataset.type == "dvc":
+        return DVCDataset(
+            type="dvc",
+            url=dataset.lock.url,
+            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 == "url":
+        fs_cls, _, path = get_cloud_fs(repo.config, url=dataset.lock.url)
+        assert fs_cls
+        join_version = getattr(fs_cls, "join_version", lambda path, _: path)
+        protocol = fs_cls.protocol
+        versioned_path = join_version(path, dataset.lock.meta.version_id)
+        versioned_path = f"{protocol}://{versioned_path}"
+        files = [
+            join_version(
+                fs_cls.join(versioned_path, file.relpath), file.meta.version_id
+            )
+            for file in dataset.lock.files
+        ]
+        return URLDataset(type="url", files=files, path=versioned_path)

dvc/api/experiments.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, Optional, Union
+from typing import Optional, Union
 
 from rich.text import Text
 
@@ -9,7 +9,7 @@ from dvc.repo.experiments.show import tabulate
 def exp_save(
     name: Optional[str] = None,
     force: bool = False,
-    include_untracked: Optional[List[str]] = None,
+    include_untracked: Optional[list[str]] = None,
 ):
     """
     Create a new DVC experiment using `exp save`.
@@ -61,12 +61,12 @@ def _postprocess(exp_rows):
 
 def exp_show(
     repo: Optional[str] = None,
-    revs: Optional[Union[str, List[str]]] = None,
+    revs: Optional[Union[str, list[str]]] = None,
     num: int = 1,
     param_deps: bool = False,
     force: bool = False,
-    config: Optional[Dict] = None,
-) -> List[Dict]:
+    config: Optional[dict] = None,
+) -> list[dict]:
     """Get DVC experiments tracked in `repo`.
 
     Without arguments, this function will retrieve all experiments derived from

dvc/api/scm.py

@@ -1,9 +1,9 @@
-from typing import List, Optional
+from typing import Optional
 
 from dvc.repo import Repo
 
 
-def all_branches(repo: Optional[str] = None) -> List[str]:
+def all_branches(repo: Optional[str] = None) -> list[str]:
     """Get all Git branches in a DVC repository.
 
     Args:
@@ -20,7 +20,7 @@ def all_branches(repo: Optional[str] = None) -> List[str]:
         return _repo.scm.list_branches()
 
 
-def all_commits(repo: Optional[str] = None) -> List[str]:
+def all_commits(repo: Optional[str] = None) -> list[str]:
     """Get all Git commits in a DVC repository.
 
     Args:
@@ -37,7 +37,7 @@ def all_commits(repo: Optional[str] = None) -> List[str]:
         return _repo.scm.list_all_commits()
 
 
-def all_tags(repo: Optional[str] = None) -> List[str]:
+def all_tags(repo: Optional[str] = None) -> list[str]:
     """Get all Git tags in a DVC repository.
 
     Args:

dvc/api/show.py

@@ -1,6 +1,7 @@
 import typing
 from collections import Counter
-from typing import Dict, Iterable, Optional, Union
+from collections.abc import Iterable
+from typing import Optional, Union
 
 from funcy import first
 
@@ -8,7 +9,7 @@ from dvc.repo import Repo
 
 
 def _postprocess(results):
-    processed: Dict[str, Dict] = {}
+    processed: dict[str, dict] = {}
     for rev, rev_data in results.items():
         if not rev_data:
             continue
@@ -24,7 +25,7 @@ def _postprocess(results):
                 (k if counts[k] == 1 else f"{file_name}:{k}"): v
                 for k, v in file_data["data"].items()
             }
-            processed[rev] = {**processed[rev], **to_merge}
+            processed[rev] = processed[rev] | to_merge
 
     if "workspace" in processed:
         del processed["workspace"]
@@ -36,8 +37,8 @@ def metrics_show(
     *targets: str,
     repo: Optional[str] = None,
     rev: Optional[str] = None,
-    config: Optional[Dict] = None,
-) -> Dict:
+    config: Optional[dict] = None,
+) -> dict:
     """Get metrics tracked in `repo`.
 
     Without arguments, this function will retrieve all metrics from all tracked
@@ -159,8 +160,8 @@ def params_show(
     stages: Optional[Union[str, Iterable[str]]] = None,
     rev: Optional[str] = None,
     deps: bool = False,
-    config: Optional[Dict] = None,
-) -> Dict:
+    config: Optional[dict] = None,
+) -> dict:
     """Get parameters tracked in `repo`.
 
     Without arguments, this function will retrieve all params from all tracked

dvc/build.py

@@ -2,7 +2,6 @@ from typing import Optional
 
 try:
     # file is created during dvc build
-    # pylint:disable=unused-import
     from . import _build  # type: ignore[attr-defined, import]
 
     PKG: Optional[str] = _build.PKG  # type: ignore[assignment]

dvc/cachemgr.py

@@ -1,5 +1,5 @@
 import os
-from typing import TYPE_CHECKING, Optional, Tuple
+from typing import TYPE_CHECKING, Optional
 
 from dvc.fs import GitFileSystem, Schemes
 from dvc_data.hashfile.db import get_odb
@@ -15,7 +15,7 @@ def _get_odb(
     repo,
     settings,
     fs=None,
-    prefix: Optional[Tuple[str, ...]] = None,
+    prefix: Optional[tuple[str, ...]] = None,
     hash_name: Optional[str] = None,
     **kwargs,
 ):
@@ -27,7 +27,7 @@ def _get_odb(
     cls, config, fs_path = get_cloud_fs(repo.config, **settings)
     fs = fs or cls(**config)
     if prefix:
-        fs_path = fs.path.join(fs_path, *prefix)
+        fs_path = fs.join(fs_path, *prefix)
     if hash_name:
         config["hash_name"] = hash_name
     return get_odb(fs, fs_path, state=repo.state, **config)
@@ -86,7 +86,7 @@ class CacheManager:
         return FileStorage(
             key=(),
             fs=self.local.fs,
-            path=self.local.fs.path.join(self.default_local_cache_dir, self.FS_DIR),
+            path=self.local.fs.join(self.default_local_cache_dir, self.FS_DIR),
         )
 
     def _init_odb(self, schemes):
@@ -137,22 +137,16 @@ def migrate_2_to_3(repo: "Repo", dry: bool = False):
     src = repo.cache.legacy
     dest = repo.cache.local
     if dry:
-        oids = list(src._list_oids())  # pylint: disable=protected-access
+        oids = list(src._list_oids())
         ui.write(
             f"{len(oids)} files will be re-hashed and migrated to the DVC 3.0 cache "
             "location."
         )
         return
 
-    with TqdmCallback(
-        desc="Computing DVC 3.0 hashes",
-        unit="files",
-    ) as cb:
+    with TqdmCallback(desc="Computing DVC 3.0 hashes", unit="files") as cb:
         migration = prepare(src, dest, callback=cb)
 
-    with TqdmCallback(
-        desc="Migrating to DVC 3.0 cache",
-        unit="files",
-    ) as cb:
+    with TqdmCallback(desc="Migrating to DVC 3.0 cache", unit="files") as cb:
         count = migrate(migration, callback=cb)
     ui.write(f"Migrated {count} files to DVC 3.0 cache location.")

dvc/cli/__init__.py

@@ -2,15 +2,17 @@
 
 import logging
 import sys
+from typing import Optional
+
+from dvc.log import logger
 
 # Workaround for CPython bug. See [1] and [2] for more info.
 # [1] https://github.com/aws/aws-cli/blob/1.16.277/awscli/clidriver.py#L55
 # [2] https://bugs.python.org/issue29288
-from typing import Optional
-
 "".encode("idna")
 
-logger = logging.getLogger("dvc")
+
+logger = logger.getChild(__name__)
 
 
 class DvcParserError(Exception):
@@ -165,9 +167,9 @@ def main(argv=None):  # noqa: C901, PLR0912, PLR0915
     # the copied parent's standard file descriptors. If we make any logging
     # calls in this state it will cause an exception due to writing to a closed
     # file descriptor.
-    if not sys.stderr or sys.stderr.closed:  # pylint: disable=using-constant-test
+    if not sys.stderr or sys.stderr.closed:
         logging.disable()
-    elif not sys.stdout or sys.stdout.closed:  # pylint: disable=using-constant-test
+    elif not sys.stdout or sys.stdout.closed:
         logging.disable(logging.INFO)
 
     args = None
@@ -197,7 +199,7 @@ def main(argv=None):  # noqa: C901, PLR0912, PLR0915
             logger.debug("v%s%s, %s on %s", __version__, pkg, pyv, platform())
             logger.debug("command: %s", " ".join(argv or sys.argv))
 
-        logger.trace(args)  # type: ignore[attr-defined]
+        logger.trace(args)
 
         if sys.stdout and not sys.stdout.closed and not args.quiet:
             from dvc.ui import ui
@@ -230,8 +232,7 @@ def main(argv=None):  # noqa: C901, PLR0912, PLR0915
         logger.exception("")
     except DvcParserError:
         ret = 254
-    except Exception as exc:  # noqa: BLE001, pylint: disable=broad-except
-        # pylint: disable=no-member
+    except Exception as exc:  # noqa: BLE001
         ret = _log_exceptions(exc) or 255
 
     try:
@@ -242,9 +243,7 @@ def main(argv=None):  # noqa: C901, PLR0912, PLR0915
         if analytics.is_enabled():
             analytics.collect_and_send_report(args, ret)
 
-        logger.trace(  # type: ignore[attr-defined]
-            "Process %s exiting with %s", os.getpid(), ret
-        )
+        logger.trace("Process %s exiting with %s", os.getpid(), ret)
 
         return ret
     finally:

dvc/cli/command.py

@@ -1,12 +1,13 @@
-import logging
 import os
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING, Any
 
+from dvc.log import logger
+
 if TYPE_CHECKING:
     from dvc.config import Config
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdBase(ABC):
@@ -31,7 +32,7 @@ class CmdBase(ABC):
 
 
 class CmdBaseNoRepo(CmdBase):
-    def __init__(self, args):  # pylint: disable=super-init-not-called
+    def __init__(self, args):
         self.args = args
 
         os.chdir(args.cd)

dvc/cli/formatter.py

@@ -0,0 +1,14 @@
+import argparse
+
+
+class HelpFormatter(argparse.HelpFormatter):
+    def _get_default_metavar_for_optional(self, action: argparse.Action) -> str:
+        return action.dest
+
+
+class RawTextHelpFormatter(HelpFormatter, argparse.RawTextHelpFormatter):
+    pass
+
+
+class RawDescriptionHelpFormatter(HelpFormatter, argparse.RawDescriptionHelpFormatter):
+    pass

dvc/cli/parser.py

@@ -1,6 +1,6 @@
 """Main parser for the dvc cli."""
+
 import argparse
-import logging
 import os
 from functools import lru_cache
 
@@ -18,8 +18,10 @@ from dvc.commands import (
     dag,
     data,
     data_sync,
+    dataset,
     destroy,
     diff,
+    du,
     experiments,
     freeze,
     gc,
@@ -27,6 +29,7 @@ from dvc.commands import (
     get_url,
     git_hook,
     imp,
+    imp_db,
     imp_url,
     init,
     install,
@@ -42,65 +45,71 @@ from dvc.commands import (
     repro,
     root,
     stage,
+    studio,
     unprotect,
     update,
     version,
 )
+from dvc.log import logger
 
-from . import DvcParserError
+from . import DvcParserError, formatter
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 COMMANDS = [
-    init,
-    queue,
-    get,
-    get_url,
-    destroy,
     add,
-    remove,
-    move,
-    unprotect,
-    repro,
-    data_sync,
-    gc,
-    imp,
-    imp_url,
-    config,
-    checkout,
-    remote,
+    artifacts,
     cache,
-    metrics,
-    params,
-    install,
-    root,
-    ls,
-    ls_url,
-    freeze,
-    dag,
-    daemon,
+    check_ignore,
+    checkout,
     commit,
     completion,
-    diff,
-    version,
-    update,
-    git_hook,
-    plots,
-    stage,
-    experiments,
-    check_ignore,
+    config,
+    daemon,
+    dag,
     data,
-    artifacts,
+    data_sync,
+    dataset,
+    destroy,
+    diff,
+    du,
+    experiments,
+    freeze,
+    gc,
+    get,
+    get_url,
+    git_hook,
+    imp,
+    imp_db,
+    imp_url,
+    init,
+    install,
+    ls,
+    ls_url,
+    metrics,
+    move,
+    params,
+    plots,
+    queue,
+    remote,
+    remove,
+    repro,
+    root,
+    stage,
+    studio,
+    unprotect,
+    update,
+    version,
 ]
 
 
 def _find_parser(parser, cmd_cls):
-    defaults = parser._defaults  # pylint: disable=protected-access
+    defaults = parser._defaults
     if not cmd_cls or cmd_cls == defaults.get("func"):
         parser.print_help()
         raise DvcParserError
 
-    actions = parser._actions  # pylint: disable=protected-access
+    actions = parser._actions
     for action in actions:
         if not isinstance(action.choices, dict):
             # NOTE: we are only interested in subparsers
@@ -112,7 +121,7 @@ def _find_parser(parser, cmd_cls):
 class DvcParser(argparse.ArgumentParser):
     """Custom parser class for dvc CLI."""
 
-    def error(self, message, cmd_cls=None):  # pylint: disable=arguments-differ
+    def error(self, message, cmd_cls=None):
         logger.error(message)
         _find_parser(self, cmd_cls)
 
@@ -159,7 +168,7 @@ def get_main_parser():
         prog="dvc",
         description=desc,
         parents=[parent_parser],
-        formatter_class=argparse.RawTextHelpFormatter,
+        formatter_class=formatter.RawTextHelpFormatter,
         add_help=False,
     )
 
@@ -194,15 +203,12 @@ def get_main_parser():
     # Sub commands
     subparsers = parser.add_subparsers(
         title="Available Commands",
-        metavar="COMMAND",
+        metavar="command",
         dest="cmd",
-        help="Use `dvc COMMAND --help` for command-specific help.",
+        help="Use `dvc command --help` for command-specific help.",
+        required=True,
     )
 
-    from .utils import fix_subparsers
-
-    fix_subparsers(subparsers)
-
     for cmd in COMMANDS:
         cmd.add_parser(subparsers, parent_parser)
 

dvc/cli/utils.py

@@ -26,18 +26,6 @@ class DictAction(argparse.Action):
         setattr(args, self.dest, d)
 
 
-def fix_subparsers(subparsers):
-    """Workaround for bug in Python 3. See more info at:
-    https://bugs.python.org/issue16308
-    https://github.com/iterative/dvc/issues/769
-
-    Args:
-        subparsers: subparsers to fix.
-    """
-    subparsers.required = True
-    subparsers.dest = "cmd"
-
-
 def append_doc_link(help_message, path):
     from dvc.utils import format_link
 

dvc/commands/add.py

@@ -1,11 +1,9 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdAdd(CmdBase):
@@ -53,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("")
@@ -71,7 +70,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(ADD_HELP, "add"),
         help=ADD_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     parser.add_argument(
         "--no-commit",
@@ -121,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

@@ -1,12 +1,10 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBaseNoRepo
-from dvc.cli.utils import DictAction, append_doc_link, fix_subparsers
+from dvc.cli.utils import DictAction, append_doc_link
 from dvc.exceptions import DvcException
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdArtifactsGet(CmdBaseNoRepo):
@@ -47,13 +45,13 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(ARTIFACTS_HELP, "artifacts"),
         help=ARTIFACTS_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     artifacts_subparsers = artifacts_parser.add_subparsers(
         dest="cmd",
         help="Use `dvc artifacts CMD --help` to display command-specific help.",
+        required=True,
     )
-    fix_subparsers(artifacts_subparsers)
 
     ARTIFACTS_GET_HELP = "Download an artifact from a DVC project."
     get_parser = artifacts_subparsers.add_parser(
@@ -61,7 +59,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(ARTIFACTS_GET_HELP, "artifacts/get"),
         help=ARTIFACTS_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     get_parser.add_argument("url", help="Location of DVC repository to download from")
     get_parser.add_argument(

dvc/commands/cache.py

@@ -1,9 +1,9 @@
 import argparse
 import os
 
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
-from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.cli.utils import append_doc_link
 from dvc.commands.config import CmdConfig
 from dvc.ui import ui
 
@@ -58,12 +58,13 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(CACHE_HELP, "cache"),
         help=CACHE_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
 
     cache_subparsers = cache_parser.add_subparsers(
         dest="cmd",
         help="Use `dvc cache CMD --help` for command-specific help.",
+        required=True,
     )
 
     parent_cache_config_parser = argparse.ArgumentParser(
@@ -76,7 +77,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser, parent_cache_config_parser],
         description=append_doc_link(CACHE_HELP, "cache/dir"),
         help=CACHE_DIR_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     cache_dir_parser.add_argument(
         "-u",
@@ -103,7 +104,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(CACHE_HELP, "cache/migrate"),
         help=CACHE_MIGRATE_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     cache_migrate_parser.add_argument(
         "--dvc-files",
@@ -122,5 +123,3 @@ def add_parser(subparsers, parent_parser):
         action="store_true",
     )
     cache_migrate_parser.set_defaults(func=CmdCacheMigrate)
-
-    fix_subparsers(cache_subparsers)

dvc/commands/check_ignore.py

@@ -1,6 +1,4 @@
-import argparse
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.ui import ui
@@ -85,7 +83,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(ADD_HELP, "check-ignore"),
         help=ADD_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     parser.add_argument(
         "-d",

dvc/commands/checkout.py

@@ -1,7 +1,6 @@
-import argparse
 import operator
 
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import CheckoutError
@@ -67,7 +66,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(CHECKOUT_HELP, "checkout"),
         help=CHECKOUT_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     checkout_parser.add_argument(
         "--summary",

dvc/commands/commit.py

@@ -1,11 +1,9 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdCommit(CmdBase):
@@ -22,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 "")
@@ -40,7 +39,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(COMMIT_HELP, "commit"),
         help=COMMIT_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     commit_parser.add_argument(
         "-f",
@@ -66,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/completion.py

@@ -1,12 +1,11 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.completion import PREAMBLE
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 SUPPORTED_SHELLS = ["bash", "zsh"]
@@ -18,7 +17,7 @@ class CmdCompletion(CmdBaseNoRepo):
 
         shell = self.args.shell
         parser = self.args.parser
-        script = shtab.complete(parser, shell=shell, preamble=PREAMBLE)  # nosec B604
+        script = shtab.complete(parser, shell=shell, preamble=PREAMBLE)
         ui.write(script, force=True)
         return 0
 
@@ -31,7 +30,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(COMPLETION_DESCRIPTION, "completion"),
         help=COMPLETION_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     completion_parser.add_argument(
         "-s",

dvc/commands/config.py

@@ -1,14 +1,17 @@
 import argparse
-import logging
 import os
 
+from funcy import set_in
+
+from dvc.cli import formatter
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
-NAME_REGEX = r"^(?P<remote>remote\.)?(?P<section>[^\.]*)\.(?P<option>[^\.]*)$"
+NAME_REGEX = r"^(?P<top>(remote|db)\.)?(?P<section>[^\.]*)\.(?P<option>[^\.]*)$"
 
 
 def _name_type(value):
@@ -17,10 +20,12 @@ def _name_type(value):
     match = re.match(NAME_REGEX, value)
     if not match:
         raise argparse.ArgumentTypeError(
-            "name argument should look like remote.name.option or section.option"
+            "name argument should look like remote.name.option or "
+            "db.name.option or section.option"
         )
+    top = match.group("top")
     return (
-        bool(match.group("remote")),
+        top.strip(".") if top else None,
         match.group("section").lower(),
         match.group("option").lower(),
     )
@@ -49,12 +54,11 @@ class CmdConfig(CmdBaseNoRepo):
             logger.error("name argument is required")
             return 1
 
-        remote, section, opt = self.args.name
+        remote_or_db, section, opt = self.args.name
 
         if self.args.value is None and not self.args.unset:
-            return self._get(remote, section, opt)
-
-        return self._set(remote, section, opt)
+            return self._get(remote_or_db, section, opt)
+        return self._set(remote_or_db, section, opt)
 
     def _list(self):
         if any((self.args.name, self.args.value, self.args.unset)):
@@ -75,18 +79,18 @@ class CmdConfig(CmdBaseNoRepo):
 
         return 0
 
-    def _get(self, remote, section, opt):
+    def _get(self, remote_or_db, section, opt):
         from dvc.config import ConfigError
 
         levels = self._get_appropriate_levels(self.args.level)[::-1]
 
         for level in levels:
             conf = self.config.read(level)
-            if remote:
-                conf = conf["remote"]
+            if remote_or_db:
+                conf = conf[remote_or_db]
 
             try:
-                self._check(conf, remote, section, opt)
+                self._check(conf, remote_or_db, section, opt)
             except ConfigError:
                 if self.args.level:
                     raise
@@ -99,16 +103,15 @@ class CmdConfig(CmdBaseNoRepo):
 
         return 0
 
-    def _set(self, remote, section, opt):
+    def _set(self, remote_or_db, section, opt):
         with self.config.edit(self.args.level) as conf:
-            if remote:
-                conf = conf["remote"]
+            if remote_or_db:
+                conf = conf[remote_or_db]
             if self.args.unset:
-                self._check(conf, remote, section, opt)
+                self._check(conf, remote_or_db, section, opt)
                 del conf[section][opt]
             else:
-                self._check(conf, remote, section)
-                conf[section][opt] = self.args.value
+                conf.update(set_in(conf, [section, opt], self.args.value))
 
         if self.args.name == "cache.type":
             logger.warning(
@@ -119,10 +122,10 @@ class CmdConfig(CmdBaseNoRepo):
 
         return 0
 
-    def _check(self, conf, remote, section, opt=None):
+    def _check(self, conf, remote_or_db, section, opt=None):
         from dvc.config import ConfigError
 
-        name = "remote" if remote else "section"
+        name = remote_or_db or "section"
         if section not in conf:
             raise ConfigError(f"{name} '{section}' doesn't exist")
 
@@ -207,7 +210,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_config_parser, parent_parser],
         description=append_doc_link(CONFIG_HELP, "config"),
         help=CONFIG_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     config_parser.add_argument(
         "-u",

dvc/commands/daemon.py

@@ -1,10 +1,8 @@
-import logging
-
 from dvc.cli import completion
 from dvc.cli.command import CmdBaseNoRepo
-from dvc.cli.utils import fix_subparsers
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdDaemonBase(CmdBaseNoRepo):
@@ -54,10 +52,9 @@ def add_parser(subparsers, parent_parser):
     daemon_subparsers = daemon_parser.add_subparsers(
         dest="cmd",
         help="Use `dvc daemon CMD --help` for command-specific help.",
+        required=True,
     )
 
-    fix_subparsers(daemon_subparsers)
-
     DAEMON_UPDATER_HELP = "Fetch latest available version."
     daemon_updater_parser = daemon_subparsers.add_parser(
         "updater",

dvc/commands/dag.py

@@ -1,6 +1,6 @@
-import argparse
 from typing import TYPE_CHECKING
 
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.ui import ui
@@ -86,7 +86,7 @@ def _collect_targets(repo, target, outs):
             targets.extend([str(out) for out in stage.outs])
             continue
 
-        for out in outs_trie.itervalues(prefix=repo.fs.path.parts(path)):  # noqa: B301
+        for out in outs_trie.itervalues(prefix=repo.fs.parts(path)):
             targets.extend(str(out))
 
     return targets
@@ -160,7 +160,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(DAG_HELP, "dag"),
         help=DAG_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     dag_parser.add_argument(
         "--dot",

dvc/commands/data.py

@@ -1,11 +1,11 @@
-import argparse
-import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, ClassVar
 
 from funcy import chunks, compact, log_durations
 
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
-from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.ui import ui
 from dvc.utils import colorize
 
@@ -13,18 +13,18 @@ if TYPE_CHECKING:
     from dvc.repo.data import Status as DataStatus
 
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdDataStatus(CmdBase):
-    COLORS = {
+    COLORS: ClassVar[dict[str, str]] = {
         "not_in_remote": "red",
         "not_in_cache": "red",
         "committed": "green",
         "uncommitted": "yellow",
         "untracked": "cyan",
     }
-    LABELS = {
+    LABELS: ClassVar[dict[str, str]] = {
         "not_in_remote": "Not in remote",
         "not_in_cache": "Not in cache",
         "committed": "DVC committed changes",
@@ -32,7 +32,7 @@ class CmdDataStatus(CmdBase):
         "untracked": "Untracked files",
         "unchanged": "DVC unchanged files",
     }
-    HINTS = {
+    HINTS: ClassVar[dict[str, tuple[str, ...]]] = {
         "not_in_remote": ('use "dvc push <file>..." to upload files',),
         "not_in_cache": ('use "dvc fetch <file>..." to download files',),
         "committed": ("git commit the corresponding dvc files to update the repo",),
@@ -106,9 +106,7 @@ class CmdDataStatus(CmdBase):
         return 0
 
     def run(self) -> int:
-        with log_durations(
-            logger.trace, "in data_status"  # type: ignore[attr-defined]
-        ):
+        with log_durations(logger.trace, "in data_status"):
             status = self.repo.data_status(
                 granular=self.args.granular,
                 untracked_files=self.args.untracked_files,
@@ -131,13 +129,13 @@ def add_parser(subparsers, parent_parser):
     data_parser = subparsers.add_parser(
         "data",
         parents=[parent_parser],
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     data_subparsers = data_parser.add_subparsers(
         dest="cmd",
         help="Use `dvc data CMD --help` to display command-specific help.",
+        required=True,
     )
-    fix_subparsers(data_subparsers)
 
     DATA_STATUS_HELP = (
         "Show changes between the last git commit, the dvcfiles and the workspace."
@@ -146,7 +144,7 @@ def add_parser(subparsers, parent_parser):
         "status",
         parents=[parent_parser],
         description=append_doc_link(DATA_STATUS_HELP, "data/status"),
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
         help=DATA_STATUS_HELP,
     )
     data_status_parser.add_argument(

dvc/commands/data_sync.py

@@ -1,11 +1,11 @@
 import argparse
-import logging
 
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdDataBase(CmdBase):
@@ -145,7 +145,7 @@ def add_parser(subparsers, _parent_parser):
         parents=[shared_parent_parser()],
         description=append_doc_link(PULL_HELP, "pull"),
         help=PULL_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     pull_parser.add_argument(
         "-r", "--remote", help="Remote storage to pull from", metavar="<name>"
@@ -220,7 +220,7 @@ def add_parser(subparsers, _parent_parser):
         parents=[shared_parent_parser()],
         description=append_doc_link(PUSH_HELP, "push"),
         help=PUSH_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     push_parser.add_argument(
         "-r", "--remote", help="Remote storage to push to", metavar="<name>"
@@ -282,7 +282,7 @@ def add_parser(subparsers, _parent_parser):
         parents=[shared_parent_parser()],
         description=append_doc_link(FETCH_HELP, "fetch"),
         help=FETCH_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     fetch_parser.add_argument(
         "-r", "--remote", help="Remote storage to fetch from", metavar="<name>"
@@ -355,7 +355,7 @@ def add_parser(subparsers, _parent_parser):
         description=append_doc_link(STATUS_HELP, "status"),
         help=STATUS_HELP,
         conflict_handler="resolve",
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     status_parser.add_argument(
         "-q",
@@ -429,5 +429,11 @@ def add_parser(subparsers, _parent_parser):
         default=False,
         help="Show status in JSON format.",
     )
+    status_parser.add_argument(
+        "--no-updates",
+        dest="check_updates",
+        action="store_false",
+        help="Ignore updates to imported data.",
+    )
 
     status_parser.set_defaults(func=CmdDataStatus)

dvc/commands/dataset.py

@@ -0,0 +1,225 @@
+from typing import TYPE_CHECKING, Optional
+
+from dvc.cli import formatter
+from dvc.cli.command import CmdBase
+from dvc.cli.utils import append_doc_link
+from dvc.exceptions import DvcException
+from dvc.log import logger
+
+if TYPE_CHECKING:
+    from rich.text import Text
+
+    from dvc.repo.datasets import Dataset, FileInfo
+
+logger = logger.getChild(__name__)
+
+
+def diff_files(old: list["FileInfo"], new: list["FileInfo"]) -> dict[str, list[str]]:
+    old_files = {d.relpath: d for d in old}
+    new_files = {d.relpath: d for d in new}
+    rest = old_files.keys() & new_files.keys()
+    return {
+        "added": list(new_files.keys() - old_files.keys()),
+        "deleted": list(old_files.keys() - new_files.keys()),
+        "modified": [p for p in rest if new_files[p] != old_files[p]],
+    }
+
+
+class CmdDatasetAdd(CmdBase):
+    @classmethod
+    def display(cls, name: str, dataset: "Dataset", action: str = "Adding"):
+        from dvc.ui import ui
+
+        assert dataset.lock
+
+        url = dataset.spec.url
+        ver: str = ""
+        if dataset.type == "dvcx":
+            ver = f"v{dataset.lock.version}"
+        if dataset.type == "dvc":
+            if dataset.lock.path:
+                url = f"{url}:/{dataset.lock.path.lstrip('/')}"
+            if rev := dataset.lock.rev:
+                ver = rev
+
+        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 "", ")")
+        ui.write(action, ui.rich_text(name, "cyan"), text, styled=True)
+
+    def run(self):
+        if not self.args.dvc and self.args.rev:
+            raise DvcException("--rev can't be used without --dvc")
+        if not self.args.dvc and self.args.path:
+            raise DvcException("--path can't be used without --dvc")
+
+        d = vars(self.args)
+        for key in ["dvc", "dvcx", "url"]:
+            if url := d.pop(key, None):
+                d.update({"type": key, "url": url})
+                break
+
+        existing = self.repo.datasets.get(self.args.name)
+        with self.repo.scm_context:
+            if not self.args.force and existing:
+                path = self.repo.fs.relpath(existing.manifest_path)
+                raise DvcException(
+                    f"{self.args.name} already exists in {path}, "
+                    "use the --force to overwrite"
+                )
+            dataset = self.repo.datasets.add(**d)
+            self.display(self.args.name, dataset)
+            return 0
+
+
+class CmdDatasetUpdate(CmdBase):
+    def display(self, name: str, dataset: "Dataset", new: "Dataset"):
+        from dvc.commands.checkout import log_changes
+        from dvc.ui import ui
+
+        action = "Updating"
+        if not dataset.lock:
+            return CmdDatasetAdd.display(name, new, action)
+        if dataset == new:
+            ui.write("[yellow]Nothing to update[/]", styled=True)
+            return
+
+        assert new.lock
+
+        v: Optional[tuple[str, str]] = None
+        if dataset.type == "dvcx":
+            assert new.type == "dvcx"
+            if new.lock.version < dataset.lock.version:
+                action = "Downgrading"
+
+            v = (f"v{dataset.lock.version}", f"v{new.lock.version}")
+        if dataset.type == "dvc":
+            assert new.type == "dvc"
+            v = (f"{dataset.lock.rev_lock[:9]}", f"{new.lock.rev_lock[:9]}")
+
+        if v:
+            part = ui.rich_text.assemble(
+                (v[0], "repr.number"),
+                " -> ",
+                (v[1], "repr.number"),
+            )
+        else:
+            part = ui.rich_text(dataset.spec.url, "repr.url")
+        changes = ui.rich_text.assemble("(", part, ")")
+        ui.write(action, ui.rich_text(name, "cyan"), changes, styled=True)
+        if dataset.type == "url":
+            assert new.type == "url"
+            stats = diff_files(dataset.lock.files, new.lock.files)
+            log_changes(stats)
+
+    def run(self):
+        from difflib import get_close_matches
+
+        from dvc.repo.datasets import DatasetNotFoundError
+        from dvc.ui import ui
+
+        version = None
+        if self.args.rev:
+            try:
+                version = int(self.args.rev.lstrip("v"))
+            except ValueError:
+                version = self.args.rev
+
+        d = vars(self.args) | {"version": version}
+        with self.repo.scm_context:
+            try:
+                dataset, new = self.repo.datasets.update(**d)
+            except DatasetNotFoundError:
+                logger.exception("")
+                if matches := get_close_matches(self.args.name, self.repo.datasets):
+                    ui.write(
+                        "did you mean?",
+                        ui.rich_text(matches[0], "cyan"),
+                        stderr=True,
+                        styled=True,
+                    )
+                return 1
+            self.display(self.args.name, dataset, new)
+            return 0
+
+
+def add_parser(subparsers, parent_parser):
+    ds_parser = subparsers.add_parser(
+        "dataset",
+        aliases=["ds"],
+        parents=[parent_parser],
+        formatter_class=formatter.RawDescriptionHelpFormatter,
+    )
+    ds_subparsers = ds_parser.add_subparsers(
+        dest="cmd",
+        help="Use `dvc dataset CMD --help` to display command-specific help.",
+        required=True,
+    )
+
+    dataset_add_help = "Add a dataset."
+    ds_add_parser = ds_subparsers.add_parser(
+        "add",
+        parents=[parent_parser],
+        description=append_doc_link(dataset_add_help, "dataset/add"),
+        formatter_class=formatter.RawTextHelpFormatter,
+        help=dataset_add_help,
+    )
+
+    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"
+    )
+    url_exclusive_group.add_argument(
+        "--dvc",
+        help="Path or URL to a Git/DVC repository to track",
+        metavar="url",
+    )
+    url_exclusive_group.add_argument(
+        "--url",
+        help="""\
+URL of a cloud-versioned remote to track. Supported URLs:
+
+s3://bucket/key/path
+gs://bucket/path/to/file/or/dir
+azure://mycontainer/path
+remote://remote_name/path/to/file/or/dir (see `dvc remote`)
+""",
+    )
+    ds_add_parser.add_argument("name", help="Name of the dataset to add")
+    ds_add_parser.add_argument(
+        "--rev",
+        help="Git revision, e.g. SHA, branch, tag (only applicable with --dvc)",
+        metavar="<commit>",
+    )
+    ds_add_parser.add_argument(
+        "--path",
+        help="Path to a file or a directory within a git repository "
+        "(only applicable with --dvc)",
+    )
+    ds_add_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Overwrite existing dataset",
+    )
+    ds_add_parser.set_defaults(func=CmdDatasetAdd)
+
+    dataset_update_help = "Update a dataset."
+    ds_update_parser = ds_subparsers.add_parser(
+        "update",
+        parents=[parent_parser],
+        description=append_doc_link(dataset_update_help, "dataset/update"),
+        formatter_class=formatter.RawDescriptionHelpFormatter,
+        help=dataset_update_help,
+    )
+    ds_update_parser.add_argument("name", help="Name of the dataset to update")
+    ds_update_parser.add_argument(
+        "--rev",
+        "--version",
+        nargs="?",
+        help="DVCX dataset version or Git revision (e.g. SHA, branch, tag)",
+        metavar="<version>",
+    )
+    ds_update_parser.set_defaults(func=CmdDatasetUpdate)

dvc/commands/destroy.py

@@ -1,10 +1,9 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdDestroy(CmdBase):
@@ -41,7 +40,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(DESTROY_HELP, "destroy"),
         help=DESTROY_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     destroy_parser.add_argument(
         "-f",

dvc/commands/diff.py

@@ -1,13 +1,12 @@
-import argparse
-import logging
 import os
 
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 def _digest(checksum):
@@ -163,7 +162,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(DIFF_DESCRIPTION, "diff"),
         help=DIFF_DESCRIPTION,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     diff_parser.add_argument(
         "--targets",

dvc/commands/du.py

@@ -0,0 +1,79 @@
+import logging
+
+from dvc.cli import completion, formatter
+from dvc.cli.command import CmdBaseNoRepo
+from dvc.cli.utils import DictAction, append_doc_link
+from dvc.ui import ui
+
+logger = logging.getLogger(__name__)
+
+
+class CmdDU(CmdBaseNoRepo):
+    def run(self):
+        from dvc.repo import Repo
+        from dvc.utils.humanize import naturalsize
+
+        entries = Repo.du(
+            self.args.url,
+            self.args.path,
+            rev=self.args.rev,
+            summarize=self.args.summarize,
+            config=self.args.config,
+            remote=self.args.remote,
+            remote_config=self.args.remote_config,
+        )
+        ui.table([(naturalsize(size), path) for path, size in entries])
+        return 0
+
+
+def add_parser(subparsers, parent_parser):
+    DU_HELP = "Show disk usage."
+    du_parser = subparsers.add_parser(
+        "du",
+        parents=[parent_parser],
+        description=append_doc_link(DU_HELP, "du"),
+        help=DU_HELP,
+        formatter_class=formatter.RawTextHelpFormatter,
+    )
+    du_parser.add_argument("url", help="Location of DVC repository")
+    du_parser.add_argument(
+        "--rev",
+        nargs="?",
+        help="Git revision (e.g. SHA, branch, tag)",
+        metavar="<commit>",
+    )
+    du_parser.add_argument(
+        "-s",
+        "--summarize",
+        action="store_true",
+        help="Show total disk usage.",
+    )
+    du_parser.add_argument(
+        "--config",
+        type=str,
+        help=(
+            "Path to a config file that will be merged with the config "
+            "in the target repository."
+        ),
+    )
+    du_parser.add_argument(
+        "--remote",
+        type=str,
+        help="Remote name to set as a default in the target repository.",
+    )
+    du_parser.add_argument(
+        "--remote-config",
+        type=str,
+        nargs="*",
+        action=DictAction,
+        help=(
+            "Remote config options to merge with a remote's config (default or one "
+            "specified by '--remote') in the target repository."
+        ),
+    )
+    du_parser.add_argument(
+        "path",
+        nargs="?",
+        help="Path to directory within the repository",
+    ).complete = completion.DIR
+    du_parser.set_defaults(func=CmdDU)

dvc/commands/experiments/__init__.py

@@ -1,6 +1,5 @@
-import argparse
-
-from dvc.cli.utils import append_doc_link, fix_subparsers, hide_subparsers_from_help
+from dvc.cli import formatter
+from dvc.cli.utils import append_doc_link, hide_subparsers_from_help
 from dvc.commands.experiments import (
     apply,
     branch,
@@ -44,16 +43,16 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         aliases=["exp"],
         description=append_doc_link(EXPERIMENTS_HELP, "exp"),
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
         help=EXPERIMENTS_HELP,
     )
 
     experiments_subparsers = experiments_parser.add_subparsers(
         dest="cmd",
         help="Use `dvc experiments CMD --help` to display command-specific help.",
+        required=True,
     )
 
-    fix_subparsers(experiments_subparsers)
     for cmd in SUB_COMMANDS:
         cmd.add_parser(experiments_subparsers, parent_parser)
     hide_subparsers_from_help(experiments_subparsers)

dvc/commands/experiments/apply.py

@@ -1,12 +1,10 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdExperimentsApply(CmdBase):
@@ -30,7 +28,7 @@ def add_parser(experiments_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(EXPERIMENTS_APPLY_HELP, "exp/apply"),
         help=EXPERIMENTS_APPLY_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     experiments_apply_parser.add_argument(
         "--no-force",

dvc/commands/experiments/branch.py

@@ -1,10 +1,9 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdExperimentsBranch(CmdBase):
@@ -21,7 +20,7 @@ def add_parser(experiments_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(EXPERIMENTS_BRANCH_HELP, "exp/branch"),
         help=EXPERIMENTS_BRANCH_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     experiments_branch_parser.add_argument(
         "experiment", help="Experiment to be promoted."

dvc/commands/experiments/clean.py

@@ -1,10 +1,9 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdExperimentsClean(CmdBase):
@@ -20,6 +19,6 @@ def add_parser(experiments_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(EXPERIMENTS_CLEAN_HELP, "exp/clean"),
         help=EXPERIMENTS_CLEAN_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     experiments_clean_parser.set_defaults(func=CmdExperimentsClean)

dvc/commands/experiments/diff.py

@@ -1,14 +1,12 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.commands.metrics import DEFAULT_PRECISION
 from dvc.exceptions import DvcException
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdExperimentsDiff(CmdBase):
@@ -59,7 +57,7 @@ def add_parser(experiments_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(EXPERIMENTS_DIFF_HELP, "exp/diff"),
         help=EXPERIMENTS_DIFF_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     experiments_diff_parser.add_argument(
         "a_rev", nargs="?", help="Old experiment to compare (defaults to HEAD)"

dvc/commands/experiments/exec_run.py

@@ -1,8 +1,7 @@
-import logging
-
 from dvc.cli.command import CmdBaseNoRepo
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdExecutorRun(CmdBaseNoRepo):

dvc/commands/experiments/ls.py

@@ -1,12 +1,11 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import InvalidArgumentError
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdExperimentsList(CmdBase):
@@ -63,7 +62,7 @@ def add_parser(experiments_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(EXPERIMENTS_LIST_HELP, "exp/list"),
         help=EXPERIMENTS_LIST_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     add_rev_selection_flags(experiments_list_parser, "List")
     display_group = experiments_list_parser.add_mutually_exclusive_group()

dvc/commands/experiments/pull.py

@@ -1,11 +1,10 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdExperimentsPull(CmdBase):
@@ -49,7 +48,7 @@ def add_parser(experiments_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(EXPERIMENTS_PULL_HELP, "exp/pull"),
         help=EXPERIMENTS_PULL_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     add_rev_selection_flags(experiments_pull_parser, "Pull", True)
     experiments_pull_parser.add_argument(

dvc/commands/experiments/push.py

@@ -1,18 +1,17 @@
-import argparse
-import logging
-from typing import Any, Dict
+from typing import Any
 
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdExperimentsPush(CmdBase):
     @staticmethod
-    def log_result(result: Dict[str, Any], remote: str):
+    def log_result(result: dict[str, Any], remote: str):
         from dvc.utils import humanize
 
         def join_exps(exps):
@@ -90,7 +89,7 @@ def add_parser(experiments_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(EXPERIMENTS_PUSH_HELP, "exp/push"),
         help=EXPERIMENTS_PUSH_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     add_rev_selection_flags(experiments_push_parser, "Push", True)
     experiments_push_parser.add_argument(

dvc/commands/experiments/queue_worker.py

@@ -1,8 +1,7 @@
-import logging
-
 from dvc.cli.command import CmdBase
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdQueueWorker(CmdBase):

dvc/commands/experiments/remove.py

@@ -1,12 +1,11 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import InvalidArgumentError
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdExperimentsRemove(CmdBase):
@@ -54,7 +53,7 @@ def add_parser(experiments_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(EXPERIMENTS_REMOVE_HELP, "exp/remove"),
         help=EXPERIMENTS_REMOVE_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     remove_group = experiments_remove_parser.add_mutually_exclusive_group()
     add_rev_selection_flags(experiments_remove_parser, "Remove", False)

dvc/commands/experiments/rename.py

@@ -1,12 +1,11 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import InvalidArgumentError
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdExperimentsRename(CmdBase):
@@ -38,7 +37,7 @@ def add_parser(experiments_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(EXPERIMENTS_RENAME_HELP, "exp/rename"),
         help=EXPERIMENTS_RENAME_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     rename_group = experiments_rename_parser.add_mutually_exclusive_group()
     rename_group.add_argument(

dvc/commands/experiments/run.py

@@ -1,11 +1,12 @@
 import argparse
-import logging
 
+from dvc.cli import formatter
 from dvc.cli.utils import append_doc_link
 from dvc.commands.repro import CmdRepro
 from dvc.commands.repro import add_arguments as add_repro_arguments
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdExperimentsRun(CmdRepro):
@@ -32,7 +33,7 @@ def add_parser(experiments_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(EXPERIMENTS_RUN_HELP, "exp/run"),
         help=EXPERIMENTS_RUN_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     _add_run_common(experiments_run_parser)
     experiments_run_parser.set_defaults(func=CmdExperimentsRun)
@@ -106,8 +107,4 @@ def _add_run_common(parser):
         default=None,
         help="Custom commit message to use when committing the experiment.",
     )
-    parser.add_argument(
-        "-M",  # obsolete
-        dest="message",
-        help=argparse.SUPPRESS,
-    )
+    parser.add_argument("-M", dest="message", help=argparse.SUPPRESS)  # obsolete

dvc/commands/experiments/save.py

@@ -1,19 +1,22 @@
 import argparse
-import logging
 
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import DvcException
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdExperimentsSave(CmdBase):
     def run(self):
         try:
             ref = self.repo.experiments.save(
+                targets=self.args.targets,
                 name=self.args.name,
+                recursive=self.args.recursive,
                 force=self.args.force,
                 include_untracked=self.args.include_untracked,
                 message=self.args.message,
@@ -38,7 +41,19 @@ def add_parser(experiments_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(EXPERIMENTS_SAVE_HELP, "exp/save"),
         help=EXPERIMENTS_SAVE_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
+    )
+    save_parser.add_argument(
+        "targets",
+        nargs="*",
+        help=("Limit DVC caching to these .dvc files and stage names."),
+    ).complete = completion.DVCFILES_AND_STAGE
+    save_parser.add_argument(
+        "-R",
+        "--recursive",
+        action="store_true",
+        default=False,
+        help="Cache subdirectories of the specified directory.",
     )
     save_parser.add_argument(
         "-f",
@@ -78,9 +93,5 @@ def add_parser(experiments_subparsers, parent_parser):
         default=None,
         help="Custom commit message to use when committing the experiment.",
     )
-    save_parser.add_argument(
-        "-M",  # obsolete
-        dest="message",
-        help=argparse.SUPPRESS,
-    )
+    save_parser.add_argument("-M", dest="message", help=argparse.SUPPRESS)  # obsolete
     save_parser.set_defaults(func=CmdExperimentsSave)

dvc/commands/experiments/show.py

@@ -1,15 +1,17 @@
 import argparse
-import logging
 import re
+from collections.abc import Iterable
 from datetime import date, datetime
-from typing import TYPE_CHECKING, Dict, Iterable
+from typing import TYPE_CHECKING
 
 from funcy import lmap
 
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.commands.metrics import DEFAULT_PRECISION
 from dvc.exceptions import DvcException
+from dvc.log import logger
 from dvc.ui import ui
 from dvc.utils.serialize import encode_exception
 
@@ -21,7 +23,7 @@ FILL_VALUE = "-"
 FILL_VALUE_ERRORED = "!"
 
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 experiment_types = {
@@ -54,9 +56,9 @@ def baseline_styler(typ):
     return {"style": "bold"} if typ == "baseline" else {}
 
 
-def show_experiments(  # noqa: C901
+def show_experiments(
     td: "TabularData",
-    headers: Dict[str, Iterable[str]],
+    headers: dict[str, Iterable[str]],
     keep=None,
     drop=None,
     pager=True,
@@ -200,7 +202,7 @@ def add_parser(experiments_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(EXPERIMENTS_SHOW_HELP, "exp/show"),
         help=EXPERIMENTS_SHOW_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     add_rev_selection_flags(experiments_show_parser, "Show")
     experiments_show_parser.add_argument(

dvc/commands/freeze.py

@@ -1,12 +1,10 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import DvcException
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdFreezeBase(CmdBase):
@@ -38,7 +36,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(FREEZE_HELP, "freeze"),
         help=FREEZE_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     freeze_parser.add_argument(
         "targets", nargs="+", help="Stages or .dvc files to freeze"
@@ -51,7 +49,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(UNFREEZE_HELP, "unfreeze"),
         help=UNFREEZE_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     unfreeze_parser.add_argument(
         "targets", nargs="+", help="Stages or .dvc files to unfreeze"

dvc/commands/gc.py

@@ -1,12 +1,12 @@
-import argparse
-import logging
 import os
 
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdGC(CmdBase):
@@ -26,6 +26,10 @@ class CmdGC(CmdBase):
             not_in_remote=self.args.not_in_remote,
         )
 
+        # Don't prompt during dry run
+        if self.args.dry:
+            self.args.force = True
+
         if self.args.rev:
             self.args.num = self.args.num or 1
 
@@ -56,7 +60,7 @@ class CmdGC(CmdBase):
             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."
 
@@ -82,6 +86,7 @@ class CmdGC(CmdBase):
             num=self.args.num,
             not_in_remote=self.args.not_in_remote,
             dry=self.args.dry,
+            skip_failed=self.args.skip_failed,
         )
         return 0
 
@@ -97,7 +102,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(GC_DESCRIPTION, "gc"),
         help=GC_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     gc_parser.add_argument(
         "-w",
@@ -151,12 +156,12 @@ def add_parser(subparsers, parent_parser):
         "--date",
         type=str,
         dest="commit_date",
-        metavar="<YYYY-MM-DD>",
+        metavar="<yyyy-mm-dd>",
         default=None,
         help=(
             "Keep cached data referenced in the commits after ( inclusive )"
             " a certain time. Date must match the extended ISO 8601 format "
-            "(YYYY-MM-DD)."
+            "(yyyy-mm-dd)."
         ),
     )
     gc_parser.add_argument(
@@ -184,6 +189,12 @@ def add_parser(subparsers, parent_parser):
         help="Remote storage to collect garbage in",
         metavar="<name>",
     )
+    gc_parser.add_argument(
+        "--skip-failed",
+        action="store_true",
+        default=False,
+        help="Skip revisions that fail when collected.",
+    )
     gc_parser.add_argument(
         "-f",
         "--force",

dvc/commands/get.py

@@ -1,12 +1,10 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.utils import DictAction, append_doc_link
 from dvc.exceptions import DvcException
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdGet(CmdBaseNoRepo):
@@ -14,7 +12,13 @@ class CmdGet(CmdBaseNoRepo):
         from dvc.api import get_url
         from dvc.ui import ui
 
-        url = get_url(self.args.path, repo=self.args.url, rev=self.args.rev)
+        url = get_url(
+            self.args.path,
+            repo=self.args.url,
+            rev=self.args.rev,
+            remote=self.args.remote,
+            remote_config=self.args.remote_config,
+        )
         ui.write(url, force=True)
 
         return 0
@@ -59,7 +63,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(GET_HELP, "get"),
         help=GET_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     get_parser.add_argument(
         "url", help="Location of DVC or Git repository to download from"

dvc/commands/get_url.py

@@ -1,12 +1,10 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.utils import DictAction, append_doc_link
 from dvc.exceptions import DvcException
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdGetUrl(CmdBaseNoRepo):
@@ -36,7 +34,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(GET_HELP, "get-url"),
         help=GET_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     get_parser.add_argument(
         "url", help="See `dvc import-url -h` for full list of supported URLs."

dvc/commands/git_hook.py

@@ -1,11 +1,11 @@
-import logging
 import os
 
+from dvc.cli import formatter
 from dvc.cli.command import CmdBaseNoRepo
-from dvc.cli.utils import fix_subparsers
 from dvc.exceptions import NotDvcRepoError
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdHookBase(CmdBaseNoRepo):
@@ -94,10 +94,9 @@ def add_parser(subparsers, parent_parser):
     git_hook_subparsers = git_hook_parser.add_subparsers(
         dest="cmd",
         help="Use `dvc daemon CMD --help` for command-specific help.",
+        required=True,
     )
 
-    fix_subparsers(git_hook_subparsers)
-
     PRE_COMMIT_HELP = "Run pre-commit GIT hook."
     pre_commit_parser = git_hook_subparsers.add_parser(
         "pre-commit",
@@ -140,6 +139,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=MERGE_DRIVER_HELP,
         help=MERGE_DRIVER_HELP,
+        formatter_class=formatter.HelpFormatter,
     )
     merge_driver_parser.add_argument(
         "--ancestor",

dvc/commands/imp.py

@@ -1,12 +1,10 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import DictAction, append_doc_link
 from dvc.exceptions import DvcException
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdImport(CmdBase):
@@ -51,7 +49,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(IMPORT_HELP, "import"),
         help=IMPORT_HELP,
-        formatter_class=argparse.RawTextHelpFormatter,
+        formatter_class=formatter.RawTextHelpFormatter,
     )
     import_parser.add_argument(
         "url", help="Location of DVC or Git repository to download from"

dvc/commands/imp_db.py

@@ -0,0 +1,100 @@
+from dvc.cli import completion, formatter
+from dvc.cli.command import CmdBase, CmdBaseNoRepo
+from dvc.cli.utils import append_doc_link
+from dvc.log import logger
+from dvc.ui import ui
+
+logger = logger.getChild(__name__)
+
+
+class CmdTestDb(CmdBaseNoRepo):
+    def run(self):
+        from dvc.config import Config
+        from dvc.database import client
+        from dvc.exceptions import DvcException
+
+        connection = self.args.conn
+        db_config = Config.from_cwd().get("db", {})
+        if connection not in db_config:
+            raise DvcException(f"connection {connection} not found in config")
+
+        config = db_config.get(connection, {})
+        if self.args.url:
+            config["url"] = self.args.url
+        if self.args.username:
+            config["username"] = self.args.username
+        if self.args.password:
+            config["password"] = self.args.password
+        with client(config) as db:
+            ui.write(f"Testing with {db}", styled=True)
+            db.test_connection()
+        ui.write("Connection successful", styled=True)
+
+
+class CmdImportDb(CmdBase):
+    def run(self):
+        self.repo.imp_db(
+            sql=self.args.sql,
+            table=self.args.table,
+            output_format=self.args.output_format,
+            out=self.args.out,
+            force=self.args.force,
+            connection=self.args.conn,
+        )
+        return 0
+
+
+def add_parser(subparsers, parent_parser):
+    IMPORT_HELP = "Snapshot a table or a SQL query result to a CSV/JSON format"
+    import_parser = subparsers.add_parser(
+        "import-db",
+        parents=[parent_parser],
+        description=append_doc_link(IMPORT_HELP, "import-db"),
+        help=IMPORT_HELP,
+        formatter_class=formatter.RawTextHelpFormatter,
+    )
+    group = import_parser.add_mutually_exclusive_group(required=True)
+    group.add_argument("--sql", help="SQL query to snapshot")
+    group.add_argument("--table", help="Table to snapshot")
+    import_parser.add_argument(
+        "--output-format",
+        default="csv",
+        const="csv",
+        nargs="?",
+        choices=["csv", "json"],
+        help="Export format",
+    )
+    import_parser.add_argument(
+        "-o",
+        "--out",
+        nargs="?",
+        help="Destination path to download files to",
+        metavar="<path>",
+    ).complete = completion.FILE
+    import_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Override destination file or folder if exists.",
+    )
+    import_parser.add_argument(
+        "--conn",
+        required=True,
+        help="Database connection to use, needs to be set in config",
+    )
+
+    import_parser.set_defaults(func=CmdImportDb)
+
+    TEST_DB_HELP = "Test the database connection"
+    test_db_parser = subparsers.add_parser(
+        "test-db",
+        parents=[parent_parser],
+        description=append_doc_link(TEST_DB_HELP, "test-db"),
+        add_help=False,
+    )
+    test_db_parser.add_argument("--conn", required=True)
+    test_db_parser.add_argument("--url")
+    test_db_parser.add_argument("--password")
+    test_db_parser.add_argument("--username")
+    test_db_parser.set_defaults(func=CmdTestDb)

dvc/commands/imp_url.py

@@ -1,12 +1,10 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import DictAction, append_doc_link
 from dvc.exceptions import DvcException
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdImportUrl(CmdBase):
@@ -44,7 +42,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(IMPORT_HELP, "import-url"),
         help=IMPORT_HELP,
-        formatter_class=argparse.RawTextHelpFormatter,
+        formatter_class=formatter.RawTextHelpFormatter,
     )
     import_parser.add_argument(
         "url",

dvc/commands/init.py

@@ -1,15 +1,14 @@
-import argparse
-import logging
-
 import colorama
 
 from dvc import analytics
+from dvc.cli import formatter
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.utils import boxify
 from dvc.utils import format_link as fmt_link
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 def _welcome_message():
@@ -69,7 +68,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(INIT_DESCRIPTION, "init"),
         help=INIT_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     init_parser.add_argument(
         "--no-scm",

dvc/commands/install.py

@@ -1,11 +1,10 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import DvcException
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdInstall(CmdBase):
@@ -25,7 +24,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(INSTALL_HELP, "install"),
         help=INSTALL_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     install_parser.add_argument(
         "--use-pre-commit-tool",

dvc/commands/ls/__init__.py

@@ -1,14 +1,12 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.utils import DictAction, append_doc_link
 from dvc.commands.ls.ls_colors import LsColors
 from dvc.exceptions import DvcException
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 def _format_entry(entry, fmt):
@@ -59,6 +57,9 @@ class CmdList(CmdBaseNoRepo):
             elif entries:
                 show_entries(entries, with_color=True, with_size=self.args.size)
             return 0
+        except FileNotFoundError:
+            logger.exception("")
+            return 1
         except DvcException:
             logger.exception("failed to list '%s'", self.args.url)
             return 1
@@ -75,7 +76,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(LIST_HELP, "list"),
         help=LIST_HELP,
-        formatter_class=argparse.RawTextHelpFormatter,
+        formatter_class=formatter.RawTextHelpFormatter,
     )
     list_parser.add_argument("url", help="Location of DVC repository to list")
     list_parser.add_argument(
@@ -121,11 +122,7 @@ def add_parser(subparsers, parent_parser):
             "specified by '--remote') in the target repository."
         ),
     )
-    list_parser.add_argument(
-        "--size",
-        action="store_true",
-        help="Show sizes.",
-    )
+    list_parser.add_argument("--size", action="store_true", help="Show sizes.")
     list_parser.add_argument(
         "path",
         nargs="?",

dvc/commands/ls/ls_colors.py

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

dvc/commands/ls_url.py

@@ -1,12 +1,11 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 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
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdListUrl(CmdBaseNoRepo):
@@ -33,22 +32,15 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(LS_HELP, "list-url"),
         help=LS_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     lsurl_parser.add_argument(
         "url", help="See `dvc import-url -h` for full list of supported URLs."
     )
     lsurl_parser.add_argument(
-        "-R",
-        "--recursive",
-        action="store_true",
-        help="Recursively list files.",
-    )
-    lsurl_parser.add_argument(
-        "--size",
-        action="store_true",
-        help="Show sizes.",
+        "-R", "--recursive", action="store_true", help="Recursively list files."
     )
+    lsurl_parser.add_argument("--size", action="store_true", help="Show sizes.")
     lsurl_parser.add_argument(
         "--fs-config",
         type=str,

dvc/commands/metrics.py

@@ -1,13 +1,11 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
-from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.ui import ui
 from dvc.utils.serialize import encode_exception
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 DEFAULT_PRECISION = 5
@@ -107,23 +105,22 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(METRICS_HELP, "metrics"),
         help=METRICS_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
 
     metrics_subparsers = metrics_parser.add_subparsers(
         dest="cmd",
         help="Use `dvc metrics CMD --help` to display command-specific help.",
+        required=True,
     )
 
-    fix_subparsers(metrics_subparsers)
-
     METRICS_SHOW_HELP = "Print metrics, with optional formatting."
     metrics_show_parser = metrics_subparsers.add_parser(
         "show",
         parents=[parent_parser],
         description=append_doc_link(METRICS_SHOW_HELP, "metrics/show"),
         help=METRICS_SHOW_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     metrics_show_parser.add_argument(
         "targets",
@@ -197,7 +194,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(METRICS_DIFF_HELP, "metrics/diff"),
         help=METRICS_DIFF_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     metrics_diff_parser.add_argument(
         "a_rev",

dvc/commands/move.py

@@ -1,12 +1,10 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import DvcException
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdMove(CmdBase):
@@ -34,7 +32,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(MOVE_DESCRIPTION, "move"),
         help=MOVE_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     move_parser.add_argument(
         "src", help="Source path to a data file or directory."

dvc/commands/params.py

@@ -1,12 +1,10 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
-from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdParamsDiff(CmdBase):
@@ -61,16 +59,15 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(PARAMS_HELP, "params"),
         help=PARAMS_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
 
     params_subparsers = params_parser.add_subparsers(
         dest="cmd",
         help="Use `dvc params CMD --help` to display command-specific help.",
+        required=True,
     )
 
-    fix_subparsers(params_subparsers)
-
     PARAMS_DIFF_HELP = (
         "Show changes in params between commits in the DVC repository, or "
         "between a commit and the workspace."
@@ -80,7 +77,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(PARAMS_DIFF_HELP, "params/diff"),
         help=PARAMS_DIFF_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     params_diff_parser.add_argument(
         "a_rev",

dvc/commands/plots.py

@@ -1,14 +1,14 @@
 import argparse
-import logging
 import os
-from typing import TYPE_CHECKING, Dict, List, Optional
+from typing import TYPE_CHECKING, Optional
 
 from funcy import compact, first, get_in
 
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
-from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.cli.utils import append_doc_link
 from dvc.exceptions import DvcException
+from dvc.log import logger
 from dvc.ui import ui
 from dvc.utils import format_link
 
@@ -16,39 +16,30 @@ if TYPE_CHECKING:
     from dvc.render.match import RendererWithErrors
 
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 def _show_json(
-    renderers_with_errors: List["RendererWithErrors"],
+    renderers_with_errors: list["RendererWithErrors"],
     split=False,
-    errors: Optional[Dict[str, Exception]] = None,
+    errors: Optional[dict[str, Exception]] = None,
 ):
     from dvc.render.convert import to_json
     from dvc.utils.serialize import encode_exception
 
-    all_errors: List[Dict] = []
+    all_errors: list[dict] = []
     data = {}
 
     for renderer, src_errors, def_errors in renderers_with_errors:
         name = renderer.name
         data[name] = to_json(renderer, split)
         all_errors.extend(
-            {
-                "name": name,
-                "rev": rev,
-                "source": source,
-                **encode_exception(e),
-            }
+            {"name": name, "rev": rev, "source": source, **encode_exception(e)}
             for rev, per_rev_src_errors in src_errors.items()
             for source, e in per_rev_src_errors.items()
         )
         all_errors.extend(
-            {
-                "name": name,
-                "rev": rev,
-                **encode_exception(e),
-            }
+            {"name": name, "rev": rev, **encode_exception(e)}
             for rev, e in def_errors.items()
         )
 
@@ -59,49 +50,6 @@ def _show_json(
     ui.write_json(compact({"errors": all_errors, "data": data}), highlight=False)
 
 
-def _adjust_vega_renderers(renderers):
-    from dvc.render import REVISION_FIELD, VERSION_FIELD
-    from dvc_render import VegaRenderer
-
-    for r in renderers:
-        if isinstance(r, VegaRenderer):
-            if _data_versions_count(r) > 1:
-                summary = _summarize_version_infos(r)
-                for dp in r.datapoints:
-                    vi = dp.pop(VERSION_FIELD, {})
-                    keys = list(vi.keys())
-                    for key in keys:
-                        if not (len(summary.get(key, set())) > 1):
-                            vi.pop(key)
-                    if vi:
-                        dp["rev"] = "::".join(vi.values())
-            else:
-                for dp in r.datapoints:
-                    dp[REVISION_FIELD] = dp[VERSION_FIELD]["revision"]
-                    dp.pop(VERSION_FIELD, {})
-
-
-def _summarize_version_infos(renderer):
-    from collections import defaultdict
-
-    from dvc.render import VERSION_FIELD
-
-    result = defaultdict(set)
-
-    for dp in renderer.datapoints:
-        for key, value in dp.get(VERSION_FIELD, {}).items():
-            result[key].add(value)
-    return dict(result)
-
-
-def _data_versions_count(renderer):
-    from itertools import product
-
-    summary = _summarize_version_infos(renderer)
-    x = product(summary.get("filename", {None}), summary.get("field", {None}))
-    return len(set(x))
-
-
 class CmdPlots(CmdBase):
     def _func(self, *args, **kwargs):
         raise NotImplementedError
@@ -144,10 +92,7 @@ class CmdPlots(CmdBase):
                 return 1
 
         try:
-            plots_data = self._func(
-                targets=self.args.targets,
-                props=self._props(),
-            )
+            plots_data = self._func(targets=self.args.targets, props=self._props())
 
             if not plots_data and not self.args.json:
                 ui.error_write(
@@ -175,11 +120,10 @@ class CmdPlots(CmdBase):
                 return 0
 
             renderers = [r.renderer for r in renderers_with_errors]
-            _adjust_vega_renderers(renderers)
             if self.args.show_vega:
                 renderer = first(filter(lambda r: r.TYPE == "vega", renderers))
                 if renderer:
-                    ui.write_json(renderer.get_filled_template(as_string=False))
+                    ui.write_json(renderer.get_filled_template())
                 return 0
 
             output_file: Path = (Path.cwd() / out).resolve() / "index.html"
@@ -267,15 +211,14 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(PLOTS_HELP, "plots"),
         help=PLOTS_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     plots_subparsers = plots_parser.add_subparsers(
         dest="cmd",
         help="Use `dvc plots CMD --help` to display command-specific help.",
+        required=True,
     )
 
-    fix_subparsers(plots_subparsers)
-
     SHOW_HELP = (
         "Generate plots from target files or from `plots` definitions in `dvc.yaml`."
     )
@@ -284,7 +227,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(SHOW_HELP, "plots/show"),
         help=SHOW_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     plots_show_parser.add_argument(
         "targets",
@@ -307,7 +250,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(PLOTS_DIFF_HELP, "plots/diff"),
         help=PLOTS_DIFF_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     plots_diff_parser.add_argument(
         "--targets",
@@ -343,7 +286,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(PLOTS_MODIFY_HELP, "plots/modify"),
         help=PLOTS_MODIFY_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     plots_modify_parser.add_argument(
         "target",
@@ -364,7 +307,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(TEMPLATES_HELP, "plots/templates"),
         help=TEMPLATES_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     plots_templates_parser.add_argument(
         "template",

dvc/commands/queue/__init__.py

@@ -1,16 +1,8 @@
-import argparse
-
-from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.cli import formatter
+from dvc.cli.utils import append_doc_link
 from dvc.commands.queue import kill, logs, remove, start, status, stop
 
-SUB_COMMANDS = [
-    start,
-    stop,
-    status,
-    logs,
-    remove,
-    kill,
-]
+SUB_COMMANDS = [start, stop, status, logs, remove, kill]
 
 
 def add_parser(subparsers, parent_parser):
@@ -20,15 +12,15 @@ def add_parser(subparsers, parent_parser):
         "queue",
         parents=[parent_parser],
         description=append_doc_link(QUEUE_HELP, "queue"),
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
         help=QUEUE_HELP,
     )
 
     queue_subparsers = queue_parser.add_subparsers(
         dest="cmd",
         help="Use `dvc queue CMD --help` to display command-specific help.",
+        required=True,
     )
 
-    fix_subparsers(queue_subparsers)
     for cmd in SUB_COMMANDS:
         cmd.add_parser(queue_subparsers, parent_parser)

dvc/commands/queue/kill.py

@@ -1,10 +1,9 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdQueueKill(CmdBase):
@@ -27,7 +26,7 @@ def add_parser(queue_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(QUEUE_KILL_HELP, "queue/kill"),
         help=QUEUE_KILL_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     queue_kill_parser.add_argument(
         "-f",

dvc/commands/queue/logs.py

@@ -1,10 +1,9 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdQueueLogs(CmdBase):
@@ -29,7 +28,7 @@ def add_parser(queue_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(QUEUE_LOGS_HELP, "queue/logs"),
         help=QUEUE_LOGS_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     queue_logs_parser.add_argument(
         "-e",
@@ -46,9 +45,5 @@ def add_parser(queue_subparsers, parent_parser):
         ),
         action="store_true",
     )
-    queue_logs_parser.add_argument(
-        "task",
-        help="Task to show.",
-        metavar="<task>",
-    )
+    queue_logs_parser.add_argument("task", help="Task to show.", metavar="<task>")
     queue_logs_parser.set_defaults(func=CmdQueueLogs)

dvc/commands/queue/remove.py

@@ -1,12 +1,11 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import InvalidArgumentError
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdQueueRemove(CmdBase):
@@ -62,7 +61,7 @@ def add_parser(queue_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(QUEUE_REMOVE_HELP, "queue/remove"),
         help=QUEUE_REMOVE_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     queue_remove_parser.add_argument(
         "--all",

dvc/commands/queue/start.py

@@ -1,11 +1,10 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdQueueStart(CmdBase):
@@ -27,7 +26,7 @@ def add_parser(queue_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(QUEUE_START_HELP, "queue/start"),
         help=QUEUE_START_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     queue_start_parser.add_argument(
         "-j",

dvc/commands/queue/status.py

@@ -1,12 +1,11 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.compare import TabularData
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdQueueStatus(CmdBase):
@@ -52,6 +51,6 @@ def add_parser(queue_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(QUEUE_STATUS_HELP, "queue/status"),
         help=QUEUE_STATUS_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     queue_status_parser.set_defaults(func=CmdQueueStatus)

dvc/commands/queue/stop.py

@@ -1,11 +1,10 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdQueueStop(CmdBase):
@@ -32,7 +31,7 @@ def add_parser(queue_subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(QUEUE_STOP_HELP, "queue/stop"),
         help=QUEUE_STOP_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     queue_stop_parser.add_argument(
         "--kill",

dvc/commands/remote.py

@@ -1,6 +1,5 @@
-import argparse
-
-from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.cli import formatter
+from dvc.cli.utils import append_doc_link
 from dvc.commands.config import CmdConfig
 from dvc.ui import ui
 from dvc.utils import format_link
@@ -155,23 +154,22 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(REMOTE_HELP, "remote"),
         help=REMOTE_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
 
     remote_subparsers = remote_parser.add_subparsers(
         dest="cmd",
         help="Use `dvc remote CMD --help` for command-specific help.",
+        required=True,
     )
 
-    fix_subparsers(remote_subparsers)
-
     REMOTE_ADD_HELP = "Add a new data remote."
     remote_add_parser = remote_subparsers.add_parser(
         "add",
         parents=[parent_config_parser, parent_parser],
         description=append_doc_link(REMOTE_ADD_HELP, "remote/add"),
         help=REMOTE_ADD_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     remote_add_parser.add_argument("name", help="Name of the remote")
     remote_add_parser.add_argument(
@@ -202,7 +200,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_config_parser, parent_parser],
         description=append_doc_link(REMOTE_DEFAULT_HELP, "remote/default"),
         help=REMOTE_DEFAULT_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     remote_default_parser.add_argument("name", nargs="?", help="Name of the remote")
     remote_default_parser.add_argument(
@@ -220,7 +218,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_config_parser, parent_parser],
         description=append_doc_link(REMOTE_MODIFY_HELP, "remote/modify"),
         help=REMOTE_MODIFY_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     remote_modify_parser.add_argument("name", help="Name of the remote")
     remote_modify_parser.add_argument("option", help="Name of the option to modify.")
@@ -242,7 +240,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_config_parser, parent_parser],
         description=append_doc_link(REMOTE_LIST_HELP, "remote/list"),
         help=REMOTE_LIST_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     remote_list_parser.set_defaults(func=CmdRemoteList)
 
@@ -252,7 +250,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_config_parser, parent_parser],
         description=append_doc_link(REMOTE_REMOVE_HELP, "remote/remove"),
         help=REMOTE_REMOVE_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     remote_remove_parser.add_argument("name", help="Name of the remote to remove.")
     remote_remove_parser.set_defaults(func=CmdRemoteRemove)
@@ -262,7 +260,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_config_parser, parent_parser],
         description=append_doc_link(REMOTE_RENAME_HELP, "remote/rename"),
         help=REMOTE_RENAME_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     remote_rename_parser.add_argument("name", help="Remote to be renamed")
     remote_rename_parser.add_argument("new", help="New name of the remote")

dvc/commands/remove.py

@@ -1,12 +1,10 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import DvcException
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdRemove(CmdBase):
@@ -30,7 +28,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(REMOVE_HELP, "remove"),
         help=REMOVE_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     remove_parser.add_argument(
         "--outs",

dvc/commands/repro.py

@@ -1,6 +1,4 @@
-import argparse
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.commands.status import CmdDataStatus
@@ -167,7 +165,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(REPRO_HELP, "repro"),
         help=REPRO_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     # repro/exp run shared args
     add_arguments(repro_parser)

dvc/commands/root.py

@@ -1,11 +1,10 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.utils import relpath
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdRoot(CmdBaseNoRepo):
@@ -24,6 +23,6 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(ROOT_HELP, "root"),
         help=ROOT_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     root_parser.set_defaults(func=CmdRoot)

dvc/commands/stage.py

@@ -1,12 +1,14 @@
 import argparse
 import logging
+from collections.abc import Iterable
 from contextlib import contextmanager
 from itertools import chain, filterfalse
-from typing import TYPE_CHECKING, Dict, Iterable, List
+from typing import TYPE_CHECKING
 
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
-from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.utils.cli_parse import parse_params
 from dvc.utils.humanize import truncate_text
 
@@ -14,7 +16,7 @@ if TYPE_CHECKING:
     from dvc.output import Output
     from dvc.stage import Stage
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 MAX_TEXT_LENGTH = 80
 
@@ -32,7 +34,7 @@ def generate_description(stage: "Stage") -> str:
     def is_plot_or_metric(out: "Output"):
         return bool(out.plot) or bool(out.metric)
 
-    desc: List[str] = []
+    desc: list[str] = []
 
     outs = list(filterfalse(is_plot_or_metric, stage.outs))
     if outs:
@@ -54,7 +56,7 @@ def prepare_stages_data(
     stages: Iterable["Stage"],
     description: bool = True,
     max_length: int = MAX_TEXT_LENGTH,
-) -> Dict[str, str]:
+) -> dict[str, str]:
     return {
         stage.addressing: (
             prepare_description(stage, max_length=max_length) if description else ""
@@ -66,10 +68,8 @@ def prepare_stages_data(
 class CmdStageList(CmdBase):
     def _get_stages(self) -> Iterable["Stage"]:
         if self.args.all:
-            stages: List["Stage"] = self.repo.index.stages
-            logger.trace(  # type: ignore[attr-defined]
-                "%d no. of stages found", len(stages)
-            )
+            stages: list["Stage"] = self.repo.index.stages
+            logger.trace("%d no. of stages found", len(stages))
             return stages
 
         # removing duplicates while maintaining order
@@ -97,7 +97,7 @@ class CmdStageList(CmdBase):
         return 0
 
 
-def parse_cmd(commands: List[str]) -> str:
+def parse_cmd(commands: list[str]) -> str:
     """
     We need to take into account two cases:
 
@@ -285,23 +285,22 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(STAGES_HELP, "stage"),
         help=STAGES_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
 
     stage_subparsers = stage_parser.add_subparsers(
         dest="cmd",
         help="Use `dvc stage CMD --help` to display command-specific help.",
+        required=True,
     )
 
-    fix_subparsers(stage_subparsers)
-
     STAGE_ADD_HELP = "Create stage"
     stage_add_parser = stage_subparsers.add_parser(
         "add",
         parents=[parent_parser],
         description=append_doc_link(STAGE_ADD_HELP, "stage/add"),
         help=STAGE_ADD_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     stage_add_parser.add_argument(
         "-n", "--name", help="Name of the stage to add", required=True
@@ -315,7 +314,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(STAGE_LIST_HELP, "stage/list"),
         help=STAGE_LIST_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     stage_list_parser.add_argument(
         "targets",

dvc/commands/status.py

@@ -1,11 +1,10 @@
-import logging
-
 from dvc.commands.data_sync import CmdDataBase
 from dvc.exceptions import DvcException
+from dvc.log import logger
 from dvc.ui import ui
 from dvc.utils import format_link
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdDataStatus(CmdDataBase):
@@ -61,18 +60,19 @@ class CmdDataStatus(CmdDataBase):
                     all_commits=self.args.all_commits,
                     with_deps=self.args.with_deps,
                     recursive=self.args.recursive,
+                    check_updates=self.args.check_updates,
                 )
             except DvcException:
                 logger.exception("")
                 return 1
 
-            if self.args.quiet:
-                return bool(st)
-
             if self.args.json:
                 ui.write_json(st)
                 return 0
 
+            if self.args.quiet:
+                return bool(st)
+
             if st:
                 self._show(st, indent)
                 return 0

dvc/commands/studio.py

@@ -0,0 +1,195 @@
+import os
+
+from funcy import get_in
+
+from dvc.cli import formatter
+from dvc.cli.utils import append_doc_link
+from dvc.commands.config import CmdConfig
+from dvc.exceptions import DvcException
+from dvc.log import logger
+
+logger = logger.getChild(__name__)
+
+
+class CmdStudioLogin(CmdConfig):
+    def run(self):
+        from dvc.env import DVC_STUDIO_URL
+        from dvc.ui import ui
+        from dvc.utils.studio import STUDIO_URL
+        from dvc_studio_client.auth import StudioAuthError, get_access_token
+
+        name = self.args.name
+        hostname = self.args.hostname or os.environ.get(DVC_STUDIO_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. "
+                "To login with a different token, "
+                "logout using 'dvc studio logout'."
+            )
+
+        open_browser = not self.args.no_open
+        try:
+            _, access_token = get_access_token(
+                token_name=name,
+                hostname=hostname,
+                scopes=scopes,
+                open_browser=open_browser,
+                client_name="DVC",
+            )
+        except StudioAuthError as e:
+            ui.error_write(str(e))
+            return 1
+
+        self.save_config(hostname, access_token)
+
+        if not self.config["exp"].get("auto_push", True):
+            from dvc.ui import ui
+
+            ui.warn(
+                "exp.auto_push is disabled. \n"
+                "Enable with 'dvc config exp.auto_push true' "
+                "to automatically push experiments to Studio."
+            )
+
+        config_path = self.config.files["global"]
+        ui.write(f"Authentication complete. Saved token to {config_path}.")
+        return 0
+
+    def save_config(self, hostname, token):
+        with self.config.edit("global") as conf:
+            conf["studio"]["token"] = token
+            conf["studio"]["url"] = hostname
+            if "auto_push" not in conf["exp"]:
+                conf["exp"]["auto_push"] = True
+
+
+class CmdStudioLogout(CmdConfig):
+    def run(self):
+        from dvc.ui import ui
+
+        with self.config.edit("global") as conf:
+            if not get_in(conf, ["studio", "token"]):
+                ui.error_write(
+                    "Not logged in to Studio. Log in with 'dvc studio login'."
+                )
+                return 1
+
+            del conf["studio"]["token"]
+
+        ui.write("Logged out from Studio (you can log back in with 'dvc studio login')")
+        return 0
+
+
+class CmdStudioToken(CmdConfig):
+    def run(self):
+        from dvc.ui import ui
+
+        conf = self.config.read("global")
+        token = get_in(conf, ["studio", "token"])
+        if not token:
+            ui.error_write("Not logged in to Studio.")
+            return 1
+
+        ui.write(token)
+        return 0
+
+
+def add_parser(subparsers, parent_parser):
+    STUDIO_HELP = "Commands to authenticate DVC with Iterative Studio"
+    STUDIO_DESCRIPTION = (
+        "Authenticate DVC with Studio and set the token."
+        " Once this token has been properly configured,\n"
+        " DVC will utilize it for seamlessly sharing live experiments\n"
+        " and sending notifications to Studio regarding any experiments"
+        " that have been pushed."
+    )
+
+    studio_parser = subparsers.add_parser(
+        "studio",
+        parents=[parent_parser],
+        description=append_doc_link(STUDIO_DESCRIPTION, "studio"),
+        help=STUDIO_HELP,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
+    )
+    studio_subparser = studio_parser.add_subparsers(
+        dest="cmd",
+        help="Use `DVC studio CMD --help` to display command-specific help.",
+        required=True,
+    )
+
+    STUDIO_LOGIN_HELP = "Authenticate DVC with Studio host"
+    STUDIO_LOGIN_DESCRIPTION = (
+        "By default, this command authenticates the DVC with Studio\n"
+        " using default scopes and assigns a random name as the token name."
+    )
+    login_parser = studio_subparser.add_parser(
+        "login",
+        parents=[parent_parser],
+        description=append_doc_link(STUDIO_LOGIN_DESCRIPTION, "studio/login"),
+        help=STUDIO_LOGIN_HELP,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
+    )
+
+    login_parser.add_argument(
+        "-H",
+        "--hostname",
+        action="store",
+        default=None,
+        help="The hostname of the Studio instance to authenticate with.",
+    )
+    login_parser.add_argument(
+        "-s",
+        "--scopes",
+        action="store",
+        default=None,
+        help="The scopes for the authentication token. ",
+    )
+
+    login_parser.add_argument(
+        "-n",
+        "--name",
+        action="store",
+        default=None,
+        help="The name of the authentication token. It will be used to\n"
+        "identify token shown in Studio profile.",
+    )
+
+    login_parser.add_argument(
+        "--no-open",
+        action="store_true",
+        default=False,
+        help="Use authentication flow based on user code.\n"
+        "You will be presented with user code to enter in browser.\n"
+        "DVC will also use this if it cannot launch browser on your behalf.",
+    )
+    login_parser.set_defaults(func=CmdStudioLogin)
+
+    STUDIO_LOGOUT_HELP = "Logout user from Studio"
+    STUDIO_LOGOUT_DESCRIPTION = (
+        "This removes the studio token from your global config.\n"
+    )
+
+    logout_parser = studio_subparser.add_parser(
+        "logout",
+        parents=[parent_parser],
+        description=append_doc_link(STUDIO_LOGOUT_DESCRIPTION, "studio/logout"),
+        help=STUDIO_LOGOUT_HELP,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
+    )
+
+    logout_parser.set_defaults(func=CmdStudioLogout)
+
+    STUDIO_TOKEN_HELP = "View the token dvc uses to contact Studio"  # noqa: S105 # nosec B105
+
+    logout_parser = studio_subparser.add_parser(
+        "token",
+        parents=[parent_parser],
+        description=append_doc_link(STUDIO_TOKEN_HELP, "studio/token"),
+        help=STUDIO_TOKEN_HELP,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
+    )
+
+    logout_parser.set_defaults(func=CmdStudioToken)

dvc/commands/unprotect.py

@@ -1,12 +1,10 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import DvcException
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdUnprotect(CmdBase):
@@ -31,7 +29,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(UNPROTECT_HELP, "unprotect"),
         help=UNPROTECT_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     unprotect_parser.add_argument(
         "targets", nargs="+", help="Data files/directories to unprotect."

dvc/commands/update.py

@@ -1,12 +1,10 @@
-import argparse
-import logging
-
-from dvc.cli import completion
+from dvc.cli import completion, formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
 from dvc.exceptions import DvcException
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdUpdate(CmdBase):
@@ -38,7 +36,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(UPDATE_HELP, "update"),
         help=UPDATE_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
     )
     update_parser.add_argument(
         "targets", nargs="+", help=".dvc files to update."

dvc/commands/version.py

@@ -1,11 +1,10 @@
-import argparse
-import logging
-
+from dvc.cli import formatter
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.utils import append_doc_link
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdVersion(CmdBaseNoRepo):
@@ -27,7 +26,7 @@ def add_parser(subparsers, parent_parser):
         parents=[parent_parser],
         description=append_doc_link(VERSION_HELP, "version"),
         help=VERSION_HELP,
-        formatter_class=argparse.RawDescriptionHelpFormatter,
+        formatter_class=formatter.RawDescriptionHelpFormatter,
         aliases=["doctor"],
     )
     version_parser.set_defaults(func=CmdVersion)

dvc/compare.py

@@ -1,23 +1,15 @@
 from collections import abc
-from itertools import chain, repeat, zip_longest
-from operator import itemgetter
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    Dict,
+from collections.abc import (
     ItemsView,
     Iterable,
     Iterator,
-    List,
     Mapping,
     MutableSequence,
-    Optional,
     Sequence,
-    Set,
-    Tuple,
-    Union,
-    overload,
 )
+from itertools import chain, repeat, zip_longest
+from operator import itemgetter
+from typing import TYPE_CHECKING, Any, Optional, Union, overload
 
 from funcy import reraise
 
@@ -25,7 +17,7 @@ if TYPE_CHECKING:
     from dvc.ui.table import CellT
 
 
-class Column(List["CellT"]):
+class Column(list["CellT"]):
     pass
 
 
@@ -35,13 +27,13 @@ def with_value(value, default):
 
 class TabularData(MutableSequence[Sequence["CellT"]]):
     def __init__(self, columns: Sequence[str], fill_value: Optional[str] = ""):
-        self._columns: Dict[str, Column] = {name: Column() for name in columns}
-        self._keys: List[str] = list(columns)
+        self._columns: dict[str, Column] = {name: Column() for name in columns}
+        self._keys: list[str] = list(columns)
         self._fill_value = fill_value
-        self._protected: Set[str] = set()
+        self._protected: set[str] = set()
 
     @property
-    def columns(self) -> List[Column]:
+    def columns(self) -> list[Column]:
         return list(map(self.column, self.keys()))
 
     def is_protected(self, col_name) -> bool:
@@ -60,10 +52,10 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
         projection = {k: self.column(k) for k in self.keys()}
         return projection.items()
 
-    def keys(self) -> List[str]:
+    def keys(self) -> list[str]:
         return self._keys
 
-    def _iter_col_row(self, row: Sequence["CellT"]) -> Iterator[Tuple["CellT", Column]]:
+    def _iter_col_row(self, row: Sequence["CellT"]) -> Iterator[tuple["CellT", Column]]:
         for val, col in zip_longest(row, self.columns):
             if col is None:
                 break
@@ -81,7 +73,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
         for val, col in self._iter_col_row(value):
             col.insert(index, val)
 
-    def __iter__(self) -> Iterator[List["CellT"]]:
+    def __iter__(self) -> Iterator[list["CellT"]]:
         return map(list, zip(*self.columns))
 
     def __getattr__(self, item: str) -> Column:
@@ -96,12 +88,10 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
         return list(it)
 
     @overload
-    def __setitem__(self, item: int, value: Sequence["CellT"]) -> None:
-        ...
+    def __setitem__(self, item: int, value: Sequence["CellT"]) -> None: ...
 
     @overload
-    def __setitem__(self, item: slice, value: Iterable[Sequence["CellT"]]) -> None:
-        ...
+    def __setitem__(self, item: slice, value: Iterable[Sequence["CellT"]]) -> None: ...
 
     def __setitem__(self, item, value) -> None:
         it = value
@@ -127,7 +117,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
         return len(self.columns[0])
 
     @property
-    def shape(self) -> Tuple[int, int]:
+    def shape(self) -> tuple[int, int]:
         return len(self.columns), len(self)
 
     def drop(self, *col_names: str) -> None:
@@ -170,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)
@@ -185,7 +175,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
 
     def as_dict(
         self, cols: Optional[Iterable[str]] = None
-    ) -> Iterable[Dict[str, "CellT"]]:
+    ) -> Iterable[dict[str, "CellT"]]:
         keys = self.keys() if cols is None else set(cols)
         return [{k: self._columns[k][i] for k in keys} for i in range(len(self))]
 
@@ -202,7 +192,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
         if how not in ["any", "all"]:
             raise ValueError(f"Invalid 'how' value {how}. Choose one of ['any', 'all']")
 
-        match_line: Set = set()
+        match_line: set = set()
         match_any = True
         if how == "all":
             match_any = False
@@ -245,7 +235,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
             )
 
         if axis == "cols":
-            cols_to_drop: List[str] = []
+            cols_to_drop: list[str] = []
             for n_col, col in enumerate(self.columns):
                 if subset and self.keys()[n_col] not in subset:
                     continue
@@ -259,7 +249,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
 
         elif axis == "rows":
             unique_rows = []
-            rows_to_drop: List[int] = []
+            rows_to_drop: list[int] = []
             for n_row, row in enumerate(self):
                 if subset:
                     row = [
@@ -319,7 +309,7 @@ def diff_table(
 ) -> TabularData:
     a_rev = a_rev or "HEAD"
     b_rev = b_rev or "workspace"
-    headers: List[str] = ["Path", title, a_rev, b_rev, "Change"]
+    headers: list[str] = ["Path", title, a_rev, b_rev, "Change"]
     fill_value = "-"
     td = TabularData(headers, fill_value=fill_value)
 
@@ -394,7 +384,7 @@ def metrics_table(
 
     for branch, val in metrics.items():
         for fname, metric in val.get("data", {}).items():
-            row_data: Dict[str, str] = {"Revision": branch, "Path": fname}
+            row_data: dict[str, str] = {"Revision": branch, "Path": fname}
             metric = metric.get("data", {})
             flattened = (
                 flatten(format_dict(metric))