[pre-commit.ci] pre-commit autoupdate (#10183)

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

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/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

@@ -1,47 +1,92 @@
-name: Build package
+name: Build and upload package
 on:
   push:
     branches:
     - main
   release:
-    types: [released, prereleased]
+    types:
+    - published
   workflow_dispatch:
 
+env:
+  FORCE_COLOR: 1
+
 permissions:
   contents: read
-  id-token: write
 
 jobs:
-  pip:
+  build:
+    name: Build & Verify package
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v4
       with:
         fetch-depth: 0
 
-    - uses: actions/setup-python@v4
+    - uses: actions/setup-python@v5
       with:
         python-version: 3.8
 
     - run: python -m pip install --upgrade pip setuptools_scm build twine
+
     - name: Force version for Test PyPI upload
       if: ${{ !startsWith(github.ref, 'refs/tags') }}
       run: echo version=$(python -m setuptools_scm | awk -F+ '{print $1}' | tail -1) >> $GITHUB_ENV
 
-    - run: |
-        echo 'PKG = "pip"'>dvc/_build.py
-        python -m build
+    - run: echo 'PKG = "pip"'>dvc/_build.py
+    - run: python -m build
       env:
-        SETUPTOOLS_SCM_PRETEND_VERSION: ${{ env.version }}
+        SETUPTOOLS_SCM_PRETEND_VERSION_FOR_DVC: ${{ env.version }}
     - run: twine check --strict dist/*
 
-    - name: Publish to PyPI
-      if: github.event_name == 'release'
-      uses: pypa/gh-action-pypi-publish@release/v1
+    - uses: actions/upload-artifact@v4
+      with:
+        name: Packages
+        path: dist/
 
-    - name: Publish to Test PyPI
-      if: ${{ github.event_name == 'release' || (github.event_name == 'push' && github.ref == 'refs/heads/main') }}
+  test-pypi-publish:
+    name: Publish dev package to test.pypi.org
+    runs-on: ubuntu-latest
+    needs: build
+    if: ${{ github.event.action == 'published' || (github.repository == 'iterative/dvc' && github.event_name == 'push' && github.ref == 'refs/heads/main') }}
+
+    environment:
+      name: test-pypi
+      url: https://test.pypi.org/p/dvc
+
+    permissions:
+      id-token: write
+
+    steps:
+    - uses: actions/download-artifact@v4
+      with:
+        name: Packages
+        path: dist
+
+    - name: Upload package to Test PyPI
       uses: pypa/gh-action-pypi-publish@release/v1
       with:
         repository-url: https://test.pypi.org/legacy/
         skip-existing: true
+
+  pypi-publish:
+    name: Publish released package to pypi.org
+    runs-on: ubuntu-latest
+    needs: build
+    if: github.event.action == 'published'
+
+    environment:
+      name: pypi
+      url: https://pypi.org/p/dvc
+
+    permissions:
+      id-token: write
+
+    steps:
+    - uses: actions/download-artifact@v4
+      with:
+        name: Packages
+        path: dist
+
+    - name: Upload package to PyPI
+      uses: pypa/gh-action-pypi-publish@release/v1

.github/workflows/codeql.yml

@@ -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

@@ -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.8", "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

.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.5.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.7.0
-    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.287'
+    rev: 'v0.1.8'
     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

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/__pyinstaller/hook-celery.py

@@ -1,5 +1,5 @@
 # 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

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

dvc/analytics.py

@@ -1,10 +1,11 @@
 import json
-import logging
 import os
 
-from .env import DVC_NO_ANALYTICS
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+from .env import DVC_ANALYTICS_ENDPOINT, DVC_NO_ANALYTICS
+
+logger = logger.getChild(__name__)
 
 
 def collect_and_send_report(args=None, return_code=None):
@@ -34,6 +35,8 @@ 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("Saving analytics report to %s", fobj.name)
     daemon(["analytics", fobj.name])
 
 
@@ -65,7 +68,7 @@ def send(path):
     """
     import requests
 
-    url = "https://analytics.dvc.org"
+    url = os.environ.get(DVC_ANALYTICS_ENDPOINT, "https://analytics.dvc.org")
     headers = {"content-type": "application/json"}
 
     with open(path, encoding="utf-8") as fobj:
@@ -73,11 +76,16 @@ def send(path):
 
     report.update(_runtime_info())
 
+    logger.debug("uploading report to %s", url)
+    logger.trace("Sending %s to %s", report, url)
+
     try:
         requests.post(url, json=report, headers=headers, timeout=5)
-    except requests.exceptions.RequestException:
-        logger.debug("failed to send analytics report", exc_info=True)
+    except requests.exceptions.RequestException as e:
+        logger.trace("", exc_info=True)
+        logger.debug("failed to send analytics report %s", str(e))
 
+    logger.trace("removing report %s", path)
     os.remove(path)
 
 

dvc/annotations.py

@@ -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,7 +1,7 @@
 from dvc.fs.dvc import _DVCFileSystem as DVCFileSystem
 
-from .data import open  # pylint: disable=redefined-builtin
-from .data import get_url, read
+from .artifacts import artifacts_show
+from .data import get_url, open, read
 from .experiments import exp_save, exp_show
 from .scm import all_branches, all_commits, all_tags
 from .show import metrics_show, params_show
@@ -10,6 +10,7 @@ __all__ = [
     "all_branches",
     "all_commits",
     "all_tags",
+    "artifacts_show",
     "exp_save",
     "exp_show",
     "get_url",

dvc/api/artifacts.py

@@ -0,0 +1,47 @@
+from typing import Any, Dict, Optional
+
+from dvc.repo import Repo
+
+
+def artifacts_show(
+    name: str,
+    version: Optional[str] = None,
+    stage: Optional[str] = None,
+    repo: Optional[str] = None,
+) -> Dict[str, str]:
+    """
+    Return path and Git revision for an artifact in a DVC project.
+
+    The resulting path and revision can be used in conjunction with other dvc.api
+    calls to open and read the artifact.
+
+    Args:
+        name (str): name of the artifact to open.
+        version (str, optional): version of the artifact to open. Defaults to
+            the latest version.
+        stage (str, optional): name of the model registry stage.
+        repo: (str, optional): path or URL for the DVC repo.
+
+    Returns:
+        Dictionary of the form:
+            {
+                "rev": ...,
+                "path": ...,
+            }
+
+    Raises:
+        dvc.exceptions.ArtifactNotFoundError: The specified artifact was not found in
+            the repo.
+    """
+    if version and stage:
+        raise ValueError("Artifact version and stage are mutually exclusive.")
+
+    repo_kwargs: Dict[str, Any] = {
+        "subrepos": True,
+        "uninitialized": True,
+    }
+    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}

dvc/api/data.py

@@ -55,7 +55,7 @@ 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,
@@ -67,7 +67,7 @@ class _OpenContextManager(GCM):
         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,

dvc/api/experiments.py

@@ -53,7 +53,7 @@ def _postprocess(exp_rows):
                 except ValueError:
                     exp_row[k] = v_str
 
-            if not exp_row[k]:
+            elif not exp_row[k]:
                 exp_row[k] = None
 
     return exp_rows

dvc/api/show.py

@@ -7,10 +7,6 @@ from funcy import first
 from dvc.repo import Repo
 
 
-def _onerror_raise(exception: Exception, *args, **kwargs):
-    raise exception
-
-
 def _postprocess(results):
     processed: Dict[str, Dict] = {}
     for rev, rev_data in results.items():
@@ -23,7 +19,6 @@ def _postprocess(results):
         for file_data in rev_data["data"].values():
             for k in file_data["data"]:
                 counts[k] += 1
-
         for file_name, file_data in rev_data["data"].items():
             to_merge = {
                 (k if counts[k] == 1 else f"{file_name}:{k}"): v
@@ -138,13 +133,17 @@ def metrics_show(
     .. _Git revision:
         https://git-scm.com/docs/revisions
     """
+    from dvc.repo.metrics.show import to_relpath
 
     with Repo.open(repo, config=config) as _repo:
         metrics = _repo.metrics.show(
             targets=targets,
             revs=rev if rev is None else [rev],
-            onerror=_onerror_raise,
+            on_error="raise",
         )
+        metrics = {
+            k: to_relpath(_repo.fs, _repo.root_dir, v) for k, v in metrics.items()
+        }
 
     metrics = _postprocess(metrics)
 
@@ -382,6 +381,8 @@ def params_show(
         https://git-scm.com/docs/revisions
 
     """
+    from dvc.repo.metrics.show import to_relpath
+
     if isinstance(stages, str):
         stages = [stages]
 
@@ -389,10 +390,11 @@ def params_show(
         params = _repo.params.show(
             revs=rev if rev is None else [rev],
             targets=targets,
-            deps=deps,
-            onerror=_onerror_raise,
+            deps_only=deps,
+            on_error="raise",
             stages=stages,
         )
+        params = {k: to_relpath(_repo.fs, _repo.root_dir, v) for k, v in params.items()}
 
     params = _postprocess(params)
 

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

@@ -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)
@@ -36,6 +36,7 @@ def _get_odb(
 class CacheManager:
     CACHE_DIR = "cache"
     FILES_DIR = "files"
+    FS_DIR = "fs"
 
     def __init__(self, repo):
         self._repo = repo
@@ -73,6 +74,21 @@ class CacheManager:
         legacy_odb = _get_odb(repo, settings, hash_name="md5-dos2unix", **kwargs)
         self._odb["legacy"] = legacy_odb
 
+    @property
+    def fs_cache(self):
+        """Filesystem-based cache.
+
+        Currently used as a temporary location to download files that we don't
+        yet have a regular oid (e.g. md5) for.
+        """
+        from dvc_data.index import FileStorage
+
+        return FileStorage(
+            key=(),
+            fs=self.local.fs,
+            path=self.local.fs.join(self.default_local_cache_dir, self.FS_DIR),
+        )
+
     def _init_odb(self, schemes):
         for scheme in schemes:
             remote = self.config.get(scheme)
@@ -121,7 +137,7 @@ 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."

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,18 +167,18 @@ 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 sys.stderr.closed:  # pylint: disable=using-constant-test
+    if not sys.stderr or sys.stderr.closed:
         logging.disable()
-    elif sys.stdout.closed:  # pylint: disable=using-constant-test
+    elif not sys.stdout or sys.stdout.closed:
         logging.disable(logging.INFO)
 
     args = None
 
     outer_log_level = logger.level
+    level = None
     try:
         args = parse_args(argv)
 
-        level = None
         if args.quiet:
             level = logging.CRITICAL
         elif args.verbose == 1:
@@ -197,9 +199,9 @@ 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 not sys.stdout.closed and not args.quiet:
+        if sys.stdout and not sys.stdout.closed and not args.quiet:
             from dvc.ui import ui
 
             ui.enable()
@@ -230,16 +232,19 @@ 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:
+        import os
+
         from dvc import analytics
 
         if analytics.is_enabled():
             analytics.collect_and_send_report(args, ret)
 
+        logger.trace("Process %s exiting with %s", os.getpid(), ret)
+
         return ret
     finally:
         logger.setLevel(outer_log_level)

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/parser.py

@@ -1,12 +1,12 @@
 """Main parser for the dvc cli."""
 import argparse
-import logging
 import os
 from functools import lru_cache
 
 from dvc import __version__
 from dvc.commands import (
     add,
+    artifacts,
     cache,
     check_ignore,
     checkout,
@@ -19,6 +19,7 @@ from dvc.commands import (
     data_sync,
     destroy,
     diff,
+    du,
     experiments,
     freeze,
     gc,
@@ -26,6 +27,7 @@ from dvc.commands import (
     get_url,
     git_hook,
     imp,
+    imp_db,
     imp_url,
     init,
     install,
@@ -41,14 +43,16 @@ from dvc.commands import (
     repro,
     root,
     stage,
+    studio,
     unprotect,
     update,
     version,
 )
+from dvc.log import logger
 
 from . import DvcParserError
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 COMMANDS = [
     init,
@@ -64,6 +68,7 @@ COMMANDS = [
     data_sync,
     gc,
     imp,
+    imp_db,
     imp_url,
     config,
     checkout,
@@ -89,16 +94,19 @@ COMMANDS = [
     experiments,
     check_ignore,
     data,
+    artifacts,
+    studio,
+    du,
 ]
 
 
 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
@@ -110,7 +118,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)
 

dvc/cli/utils.py

@@ -2,6 +2,10 @@ import argparse
 
 
 class DictAction(argparse.Action):
+    def __init__(self, *args, **kwargs):
+        kwargs.setdefault("metavar", "<name>=<value>")
+        super().__init__(*args, **kwargs)
+
     def __call__(self, parser, args, values, option_string=None):  # noqa: ARG002
         d = getattr(args, self.dest) or {}
 

dvc/commands/add.py

@@ -1,11 +1,11 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):

dvc/commands/artifacts.py

@@ -0,0 +1,133 @@
+import argparse
+
+from dvc.cli import completion
+from dvc.cli.command import CmdBaseNoRepo
+from dvc.cli.utils import DictAction, append_doc_link, fix_subparsers
+from dvc.exceptions import DvcException
+from dvc.log import logger
+
+logger = logger.getChild(__name__)
+
+
+class CmdArtifactsGet(CmdBaseNoRepo):
+    def run(self):
+        from dvc.repo.artifacts import Artifacts
+        from dvc.scm import CloneError
+        from dvc.ui import ui
+
+        try:
+            count, out = Artifacts.get(
+                self.args.url,
+                name=self.args.name,
+                version=self.args.rev,
+                stage=self.args.stage,
+                force=self.args.force,
+                config=self.args.config,
+                remote=self.args.remote,
+                remote_config=self.args.remote_config,
+                out=self.args.out,
+            )
+            ui.write(f"Downloaded {count} file(s) to '{out}'")
+            return 0
+        except CloneError:
+            logger.exception("failed to get '%s'", self.args.name)
+            return 1
+        except DvcException:
+            logger.exception(
+                "failed to get '%s' from '%s'", self.args.name, self.args.url
+            )
+            return 1
+
+
+def add_parser(subparsers, parent_parser):
+    ARTIFACTS_HELP = "DVC model registry artifact commands."
+
+    artifacts_parser = subparsers.add_parser(
+        "artifacts",
+        parents=[parent_parser],
+        description=append_doc_link(ARTIFACTS_HELP, "artifacts"),
+        help=ARTIFACTS_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    artifacts_subparsers = artifacts_parser.add_subparsers(
+        dest="cmd",
+        help="Use `dvc artifacts CMD --help` to display command-specific help.",
+    )
+    fix_subparsers(artifacts_subparsers)
+
+    ARTIFACTS_GET_HELP = "Download an artifact from a DVC project."
+    get_parser = artifacts_subparsers.add_parser(
+        "get",
+        parents=[parent_parser],
+        description=append_doc_link(ARTIFACTS_GET_HELP, "artifacts/get"),
+        help=ARTIFACTS_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    get_parser.add_argument("url", help="Location of DVC repository to download from")
+    get_parser.add_argument(
+        "name", help="Name of artifact in the repository"
+    ).complete = completion.FILE
+    get_parser.add_argument(
+        "--rev",
+        nargs="?",
+        help="Artifact version",
+        metavar="<version>",
+    )
+    get_parser.add_argument(
+        "--stage",
+        nargs="?",
+        help="Artifact stage",
+        metavar="<stage>",
+    )
+    get_parser.add_argument(
+        "-o",
+        "--out",
+        nargs="?",
+        help="Destination path to download artifact to",
+        metavar="<path>",
+    ).complete = completion.DIR
+    get_parser.add_argument(
+        "-j",
+        "--jobs",
+        type=int,
+        help=(
+            "Number of jobs to run simultaneously. "
+            "The default value is 4 * cpu_count(). "
+        ),
+        metavar="<number>",
+    )
+    get_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Override local file or folder if exists.",
+    )
+    get_parser.add_argument(
+        "--config",
+        type=str,
+        help=(
+            "Path to a config file that will be merged with the config "
+            "in the target repository."
+        ),
+    )
+    get_parser.add_argument(
+        "--remote",
+        type=str,
+        help=(
+            "Remote name to set as a default in the target repository "
+            "(only applicable when downloading from DVC remote)."
+        ),
+    )
+    get_parser.add_argument(
+        "--remote-config",
+        type=str,
+        nargs="*",
+        action=DictAction,
+        help=(
+            "Remote config options to merge with a remote's config (default or one "
+            "specified by '--remote') in the target repository (only applicable "
+            "when downloading from DVC remote)."
+        ),
+    )
+    get_parser.set_defaults(func=CmdArtifactsGet)

dvc/commands/cache.py

@@ -40,8 +40,11 @@ class CmdCacheDir(CmdConfig):
 class CmdCacheMigrate(CmdBase):
     def run(self):
         from dvc.cachemgr import migrate_2_to_3
+        from dvc.repo.commit import commit_2_to_3
 
         migrate_2_to_3(self.repo, dry=self.args.dry)
+        if self.args.dvc_files:
+            commit_2_to_3(self.repo, dry=self.args.dry)
         return 0
 
 
@@ -102,6 +105,14 @@ def add_parser(subparsers, parent_parser):
         help=CACHE_MIGRATE_HELP,
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )
+    cache_migrate_parser.add_argument(
+        "--dvc-files",
+        help=(
+            "Migrate entries in all existing DVC files in the repository "
+            "to the DVC 3.0 format."
+        ),
+        action="store_true",
+    )
     cache_migrate_parser.add_argument(
         "--dry",
         help=(

dvc/commands/commit.py

@@ -1,11 +1,11 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):

dvc/commands/completion.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 
 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 +18,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
 

dvc/commands/config.py

@@ -1,14 +1,16 @@
 import argparse
-import logging
 import os
 
+from funcy import set_in
+
 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 +19,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 +53,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 +78,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 +102,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 +121,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")
 

dvc/commands/daemon.py

@@ -1,6 +1,9 @@
 from dvc.cli import completion
 from dvc.cli.command import CmdBaseNoRepo
 from dvc.cli.utils import fix_subparsers
+from dvc.log import logger
+
+logger = logger.getChild(__name__)
 
 
 class CmdDaemonBase(CmdBaseNoRepo):
@@ -21,6 +24,8 @@ class CmdDaemonUpdater(CmdDaemonBase):
         config = Config(dvc_dir, validate=False)
         hardlink_lock = config.get("core", {}).get("hardlink_lock", False)
         updater = Updater(tmp_dir, hardlink_lock=hardlink_lock)
+
+        logger.info("Starting updater to fetch the latest version")
         updater.fetch(detach=False)
 
         return 0
@@ -30,6 +35,7 @@ class CmdDaemonAnalytics(CmdDaemonBase):
     def run(self):
         from dvc import analytics
 
+        logger.info("Sending analytics")
         analytics.send(self.args.target)
 
         return 0

dvc/commands/dag.py

@@ -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

dvc/commands/data.py

@@ -1,11 +1,11 @@
 import argparse
-import logging
 from typing import TYPE_CHECKING
 
 from funcy import chunks, compact, log_durations
 
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.log import logger
 from dvc.ui import ui
 from dvc.utils import colorize
 
@@ -13,7 +13,7 @@ if TYPE_CHECKING:
     from dvc.repo.data import Status as DataStatus
 
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdDataStatus(CmdBase):
@@ -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,

dvc/commands/data_sync.py

@@ -1,11 +1,11 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):
@@ -14,6 +14,10 @@ class CmdDataBase(CmdBase):
         from dvc.utils.humanize import get_summary
 
         default_msg = "Everything is up to date."
+
+        if not self.args.remote and not self.repo.config["core"].get("remote"):
+            ui.warn("No remote provided and no default remote set.")
+
         ui.write(get_summary(stats.items()) or default_msg)
 
 

dvc/commands/destroy.py

@@ -1,10 +1,10 @@
 import argparse
-import logging
 
 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):

dvc/commands/diff.py

@@ -1,13 +1,13 @@
 import argparse
-import logging
 import os
 
 from dvc.cli import completion
 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):

dvc/commands/du.py

@@ -0,0 +1,80 @@
+import argparse
+import logging
+
+from dvc.cli import completion
+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=argparse.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

@@ -12,6 +12,7 @@ from dvc.commands.experiments import (
     push,
     queue_worker,
     remove,
+    rename,
     run,
     save,
     show,
@@ -28,6 +29,7 @@ SUB_COMMANDS = [
     push,
     queue_worker,
     remove,
+    rename,
     run,
     save,
     show,

dvc/commands/experiments/apply.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):

dvc/commands/experiments/branch.py

@@ -1,10 +1,10 @@
 import argparse
-import logging
 
 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):

dvc/commands/experiments/clean.py

@@ -1,10 +1,10 @@
 import argparse
-import logging
 
 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):

dvc/commands/experiments/diff.py

@@ -1,14 +1,14 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):

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):
@@ -49,7 +48,7 @@ def add_parser(experiments_subparsers, parent_parser):
         ),
     )
     exec_run_parser.add_argument(
-        "-M",
+        "-m",
         "--message",
         type=str,
         default=None,

dvc/commands/experiments/ls.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 
 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):

dvc/commands/experiments/pull.py

@@ -1,11 +1,11 @@
 import argparse
-import logging
 
 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):

dvc/commands/experiments/push.py

@@ -1,13 +1,13 @@
 import argparse
-import logging
 from typing import Any, Dict
 
 from dvc.cli import completion
 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):

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,12 @@
 import argparse
-import logging
 
 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):

dvc/commands/experiments/rename.py

@@ -0,0 +1,68 @@
+import argparse
+
+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 = logger.getChild(__name__)
+
+
+class CmdExperimentsRename(CmdBase):
+    def run(self):
+        from dvc.utils import humanize
+
+        if not (self.args.experiment and self.args.name):
+            raise InvalidArgumentError(
+                "An experiment to rename and a new experiment name are required."
+            )
+        renamed = self.repo.experiments.rename(
+            exp_name=self.args.experiment,
+            new_name=self.args.name,
+            git_remote=self.args.git_remote,
+            force=self.args.force,
+        )
+        if renamed:
+            ui.write(f"Renamed experiments: {humanize.join(map(repr, renamed))}")
+        else:
+            ui.write("No experiments to rename.")
+
+        return 0
+
+
+def add_parser(experiments_subparsers, parent_parser):
+    EXPERIMENTS_RENAME_HELP = "Rename experiments."
+    experiments_rename_parser = experiments_subparsers.add_parser(
+        "rename",
+        parents=[parent_parser],
+        description=append_doc_link(EXPERIMENTS_RENAME_HELP, "exp/rename"),
+        help=EXPERIMENTS_RENAME_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    rename_group = experiments_rename_parser.add_mutually_exclusive_group()
+    rename_group.add_argument(
+        "-g",
+        "--git-remote",
+        metavar="<git_remote>",
+        help="Name or URL of the Git remote to rename the experiment from",
+    )
+    experiments_rename_parser.add_argument(
+        "experiment",
+        help="Experiment to rename.",
+        nargs="?",
+        metavar="<experiment>",
+    )
+    experiments_rename_parser.add_argument(
+        "name",
+        help="New name for the experiment.",
+        metavar="<name>",
+    )
+    experiments_rename_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Replace experiment if it already exists.",
+    )
+    experiments_rename_parser.set_defaults(func=CmdExperimentsRename)

dvc/commands/experiments/run.py

@@ -1,11 +1,11 @@
 import argparse
-import logging
 
 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):
@@ -26,7 +26,7 @@ class CmdExperimentsRun(CmdRepro):
 
 
 def add_parser(experiments_subparsers, parent_parser):
-    EXPERIMENTS_RUN_HELP = "Run or resume an experiment."
+    EXPERIMENTS_RUN_HELP = "Run an experiment."
     experiments_run_parser = experiments_subparsers.add_parser(
         "run",
         parents=[parent_parser],
@@ -106,3 +106,8 @@ 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,
+    )

dvc/commands/experiments/save.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 
 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):
@@ -72,10 +72,15 @@ def add_parser(experiments_subparsers, parent_parser):
         metavar="<path>",
     )
     save_parser.add_argument(
-        "-M",
+        "-m",
         "--message",
         type=str,
         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.set_defaults(func=CmdExperimentsSave)

dvc/commands/experiments/show.py

@@ -1,5 +1,4 @@
 import argparse
-import logging
 import re
 from datetime import date, datetime
 from typing import TYPE_CHECKING, Dict, Iterable
@@ -10,6 +9,7 @@ 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 +21,7 @@ FILL_VALUE = "-"
 FILL_VALUE_ERRORED = "!"
 
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 experiment_types = {
@@ -54,7 +54,7 @@ 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]],
     keep=None,

dvc/commands/freeze.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):

dvc/commands/gc.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 import os
 
 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
 
@@ -81,6 +85,7 @@ class CmdGC(CmdBase):
             rev=self.args.rev,
             num=self.args.num,
             not_in_remote=self.args.not_in_remote,
+            dry=self.args.dry,
         )
         return 0
 
@@ -213,4 +218,10 @@ def add_parser(subparsers, parent_parser):
         ),
         metavar="<paths>",
     )
+    gc_parser.add_argument(
+        "--dry",
+        action="store_true",
+        default=False,
+        help=("Only print what would get removed without actually removing."),
+    )
     gc_parser.set_defaults(func=CmdGC)

dvc/commands/get.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):

dvc/commands/get_url.py

@@ -1,16 +1,17 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):
     def run(self):
+        from dvc.config import Config
         from dvc.repo import Repo
 
         try:
@@ -20,6 +21,7 @@ class CmdGetUrl(CmdBaseNoRepo):
                 jobs=self.args.jobs,
                 force=self.args.force,
                 fs_config=self.args.fs_config,
+                config=Config.from_cwd(),
             )
             return 0
         except DvcException:

dvc/commands/git_hook.py

@@ -1,11 +1,11 @@
-import logging
 import os
 
 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):

dvc/commands/imp.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):
@@ -25,6 +25,7 @@ class CmdImport(CmdBase):
                 config=self.args.config,
                 remote=self.args.remote,
                 remote_config=self.args.remote_config,
+                force=self.args.force,
             )
         except CloneError:
             logger.exception("failed to import '%s'", self.args.path)
@@ -65,6 +66,13 @@ def add_parser(subparsers, parent_parser):
         help="Destination path to download files to",
         metavar="<path>",
     ).complete = completion.DIR
+    import_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Override destination file or folder if exists.",
+    )
     import_parser.add_argument(
         "--rev",
         nargs="?",

dvc/commands/imp_db.py

@@ -0,0 +1,167 @@
+import argparse
+
+from funcy import compact, merge
+
+from dvc.cli import completion
+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 = logger.getChild(__name__)
+
+
+class CmdTestDb(CmdBase):
+    def run(self):
+        from dvc.database import get_client
+        from dvc.database.dbt_utils import DBT_PROJECT_FILE, is_dbt_project
+        from dvc.dependency.db import _get_dbt_config
+        from dvc.exceptions import DvcException
+
+        connection = self.args.connection
+
+        db_config = self.repo.config.get("db", {})
+        config = db_config.get(connection, {})
+        if connection and not config:
+            raise DvcException(f"connection {connection} not found in config")
+
+        cli_config = compact(
+            {
+                "url": self.args.url,
+                "username": self.args.username,
+                "password": self.args.password,
+            }
+        )
+        conn_config = merge(config, cli_config)
+
+        cli_dbt_config = {"profile": self.args.profile, "target": self.args.target}
+        dbt_config = merge(_get_dbt_config(self.repo.config), cli_dbt_config)
+
+        project_dir = self.repo.root_dir
+        if not (conn_config or dbt_config):
+            if is_dbt_project(project_dir):
+                ui.write("Using", DBT_PROJECT_FILE, "for testing", styled=True)
+            else:
+                raise DvcException(
+                    f"no config set and {DBT_PROJECT_FILE} is missing; "
+                    "provide arguments or set a configuration"
+                )
+
+        adapter = get_client(conn_config, project_dir=project_dir, **dbt_config)
+        with adapter as db:
+            ui.write(f"Testing with {db}", styled=True)
+
+            creds = getattr(db, "creds", {})
+            for k, v in creds.items():
+                ui.write("\t", f"{k}:", v, styled=True)
+
+            if creds:
+                ui.write()
+
+            db.test_connection()
+
+        ui.write("Connection successful", styled=True)
+
+
+class CmdImportDb(CmdBase):
+    def run(self):
+        if not (self.args.sql or self.args.model):
+            raise argparse.ArgumentTypeError("Either of --sql or --model is required.")
+
+        self.repo.imp_db(
+            url=self.args.url,
+            rev=self.args.rev,
+            project_dir=self.args.project_dir,
+            sql=self.args.sql,
+            model=self.args.model,
+            profile=self.args.profile,
+            target=self.args.target,
+            output_format=self.args.output_format,
+            out=self.args.out,
+            force=self.args.force,
+            connection=self.args.connection,
+        )
+        return 0
+
+
+def add_parser(subparsers, parent_parser):
+    IMPORT_HELP = (
+        "Download file or directory tracked by DVC or by Git "
+        "into the workspace, and track it."
+    )
+
+    import_parser = subparsers.add_parser(
+        "import-db",
+        parents=[parent_parser],
+        description=append_doc_link(IMPORT_HELP, "import"),
+        add_help=False,
+    )
+    import_parser.add_argument(
+        "--url", help="Location of DVC or Git repository to download from"
+    )
+    import_parser.add_argument(
+        "--rev",
+        nargs="?",
+        help="Git revision (e.g. SHA, branch, tag)",
+        metavar="<commit>",
+    )
+    import_parser.add_argument(
+        "--project-dir", nargs="?", help="Subdirectory to the dbt project location"
+    )
+
+    group = import_parser.add_mutually_exclusive_group()
+    group.add_argument(
+        "--sql",
+        help="SQL query",
+    )
+    group.add_argument(
+        "--model",
+        help="Model name to download",
+    )
+    import_parser.add_argument("--profile", help="Profile to use")
+    import_parser.add_argument("--target", help="Target to use")
+    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",
+        dest="connection",
+        nargs="?",
+        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", dest="connection")
+    test_db_parser.add_argument("--url")
+    test_db_parser.add_argument("--password")
+    test_db_parser.add_argument("--username")
+    test_db_parser.add_argument("--profile")
+    test_db_parser.add_argument("--target")
+    test_db_parser.set_defaults(func=CmdTestDb)

dvc/commands/imp_url.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):

dvc/commands/init.py

@@ -1,15 +1,15 @@
 import argparse
-import logging
 
 import colorama
 
 from dvc import analytics
 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():

dvc/commands/install.py

@@ -1,11 +1,11 @@
 import argparse
-import logging
 
 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):

dvc/commands/ls/__init__.py

@@ -1,14 +1,14 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):

dvc/commands/ls_url.py

@@ -1,22 +1,24 @@
 import argparse
-import logging
 
 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):
     def run(self):
+        from dvc.config import Config
         from dvc.repo import Repo
 
         entries = Repo.ls_url(
             self.args.url,
             recursive=self.args.recursive,
             fs_config=self.args.fs_config,
+            config=Config.from_cwd(),
         )
         if entries:
             show_entries(entries, with_color=True, with_size=self.args.size)

dvc/commands/metrics.py

@@ -1,14 +1,13 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link, fix_subparsers
-from dvc.exceptions import DvcException
+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
@@ -20,17 +19,25 @@ class CmdMetricsBase(CmdBase):
 
 class CmdMetricsShow(CmdMetricsBase):
     def run(self):
-        try:
-            metrics = self.repo.metrics.show(
-                self.args.targets,
-                all_branches=self.args.all_branches,
-                all_tags=self.args.all_tags,
-                all_commits=self.args.all_commits,
-                recursive=self.args.recursive,
+        from dvc.repo.metrics.show import to_relpath
+        from dvc.utils import errored_revisions
+
+        metrics = self.repo.metrics.show(
+            self.args.targets,
+            all_branches=self.args.all_branches,
+            all_tags=self.args.all_tags,
+            all_commits=self.args.all_commits,
+        )
+        metrics = {
+            k: to_relpath(self.repo.fs, self.repo.root_dir, v)
+            for k, v in metrics.items()
+        }
+
+        if errored := errored_revisions(metrics):
+            ui.error_write(
+                "DVC failed to load some metrics for following revisions:"
+                f" '{', '.join(errored)}'."
             )
-        except DvcException:
-            logger.exception("")
-            return 1
 
         if self.args.json:
             ui.write_json(metrics, default=encode_exception)
@@ -52,17 +59,26 @@ class CmdMetricsShow(CmdMetricsBase):
 
 class CmdMetricsDiff(CmdMetricsBase):
     def run(self):
-        try:
-            diff = self.repo.metrics.diff(
-                a_rev=self.args.a_rev,
-                b_rev=self.args.b_rev,
-                targets=self.args.targets,
-                recursive=self.args.recursive,
-                all=self.args.all,
+        import os
+        from os.path import relpath
+
+        diff_result = self.repo.metrics.diff(
+            a_rev=self.args.a_rev,
+            b_rev=self.args.b_rev,
+            targets=self.args.targets,
+            all=self.args.all,
+        )
+
+        errored = [rev for rev, err in diff_result.get("errors", {}).items() if err]
+        if errored:
+            ui.error_write(
+                "DVC failed to load some metrics for following revisions:"
+                f" '{', '.join(errored)}'."
             )
-        except DvcException:
-            logger.exception("failed to show metrics diff")
-            return 1
+
+        start = relpath(os.getcwd(), self.repo.root_dir)
+        diff = diff_result.get("diff", {})
+        diff = {relpath(path, start): result for path, result in diff.items()}
 
         if self.args.json:
             ui.write_json(diff)
@@ -184,10 +200,14 @@ def add_parser(subparsers, parent_parser):
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )
     metrics_diff_parser.add_argument(
-        "a_rev", nargs="?", help="Old Git commit to compare (defaults to HEAD)"
+        "a_rev",
+        nargs="?",
+        help="Old Git commit to compare (defaults to HEAD)",
+        default="HEAD",
     )
     metrics_diff_parser.add_argument(
         "b_rev",
+        default="workspace",
         nargs="?",
         help="New Git commit to compare (defaults to the current workspace)",
     )

dvc/commands/move.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):

dvc/commands/params.py

@@ -1,30 +1,39 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link, fix_subparsers
-from dvc.exceptions import DvcException
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdParamsDiff(CmdBase):
     UNINITIALIZED = True
 
     def run(self):
-        try:
-            diff = self.repo.params.diff(
-                a_rev=self.args.a_rev,
-                b_rev=self.args.b_rev,
-                targets=self.args.targets,
-                all=self.args.all,
-                deps=self.args.deps,
+        import os
+        from os.path import relpath
+
+        diff_result = self.repo.params.diff(
+            a_rev=self.args.a_rev,
+            b_rev=self.args.b_rev,
+            targets=self.args.targets,
+            all=self.args.all,
+            deps_only=self.args.deps,
+        )
+
+        errored = [rev for rev, err in diff_result.get("errors", {}).items() if err]
+        if errored:
+            ui.error_write(
+                "DVC failed to load some metrics for following revisions:"
+                f" '{', '.join(errored)}'."
             )
-        except DvcException:
-            logger.exception("failed to show params diff")
-            return 1
+
+        start = relpath(os.getcwd(), self.repo.root_dir)
+        diff = diff_result.get("diff", {})
+        diff = {relpath(path, start): result for path, result in diff.items()}
 
         if self.args.json:
             ui.write_json(diff)
@@ -74,10 +83,14 @@ def add_parser(subparsers, parent_parser):
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )
     params_diff_parser.add_argument(
-        "a_rev", nargs="?", help="Old Git commit to compare (defaults to HEAD)"
+        "a_rev",
+        nargs="?",
+        default="HEAD",
+        help="Old Git commit to compare (defaults to HEAD)",
     )
     params_diff_parser.add_argument(
         "b_rev",
+        default="workspace",
         nargs="?",
         help="New Git commit to compare (defaults to the current workspace)",
     )

dvc/commands/plots.py

@@ -1,5 +1,4 @@
 import argparse
-import logging
 import os
 from typing import TYPE_CHECKING, Dict, List, Optional
 
@@ -9,6 +8,7 @@ from dvc.cli import completion
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link, fix_subparsers
 from dvc.exceptions import DvcException
+from dvc.log import logger
 from dvc.ui import ui
 from dvc.utils import format_link
 
@@ -16,7 +16,7 @@ if TYPE_CHECKING:
     from dvc.render.match import RendererWithErrors
 
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 def _show_json(
@@ -59,49 +59,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
@@ -175,11 +132,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"

dvc/commands/queue/kill.py

@@ -1,10 +1,10 @@
 import argparse
-import logging
 
 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):

dvc/commands/queue/logs.py

@@ -1,10 +1,10 @@
 import argparse
-import logging
 
 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):

dvc/commands/queue/remove.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 
 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):

dvc/commands/queue/start.py

@@ -1,11 +1,11 @@
 import argparse
-import logging
 
 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):

dvc/commands/queue/status.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 
 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):

dvc/commands/queue/stop.py

@@ -1,11 +1,11 @@
 import argparse
-import logging
 
 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):

dvc/commands/remove.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):

dvc/commands/root.py

@@ -1,11 +1,11 @@
 import argparse
-import logging
 
 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):

dvc/commands/stage.py

@@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Dict, Iterable, List
 from dvc.cli import completion
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.log import logger
 from dvc.utils.cli_parse import parse_params
 from dvc.utils.humanize import truncate_text
 
@@ -14,7 +15,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
 
@@ -67,9 +68,7 @@ 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)
-            )
+            logger.trace("%d no. of stages found", len(stages))
             return stages
 
         # removing duplicates while maintaining order

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):

dvc/commands/studio.py

@@ -0,0 +1,175 @@
+import argparse
+import os
+
+from funcy import get_in
+
+from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.commands.config import CmdConfig
+from dvc.log import logger
+
+logger = logger.getChild(__name__)
+
+
+class CmdStudioLogin(CmdConfig):
+    def run(self):
+        from dvc_studio_client.auth import StudioAuthError, get_access_token
+
+        from dvc.env import DVC_STUDIO_URL
+        from dvc.ui import ui
+        from dvc.utils.studio import STUDIO_URL
+
+        name = self.args.name
+        hostname = self.args.hostname or os.environ.get(DVC_STUDIO_URL) or STUDIO_URL
+        scopes = self.args.scopes
+
+        try:
+            token_name, access_token = get_access_token(
+                token_name=name,
+                hostname=hostname,
+                scopes=scopes,
+                use_device_code=self.args.no_open,
+                client_name="DVC",
+            )
+        except StudioAuthError as e:
+            ui.error_write(str(e))
+            return 1
+
+        self.save_config(hostname, access_token)
+        ui.write(
+            "Authentication has been successfully completed."
+            "The generated token will now be accessible as"
+            f" {token_name} in the user's Studio profile."
+        )
+        return 0
+
+    def save_config(self, hostname, token):
+        with self.config.edit("global") as conf:
+            conf["studio"]["token"] = token
+            conf["studio"]["url"] = hostname
+
+
+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.")
+                return 1
+
+            del conf["studio"]["token"]
+
+        ui.write("Logged out from Studio")
+        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=argparse.RawDescriptionHelpFormatter,
+    )
+    studio_subparser = studio_parser.add_subparsers(
+        dest="cmd",
+        help="Use `DVC studio CMD --help` to display command-specific help.",
+    )
+    fix_subparsers(studio_subparser)
+
+    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=argparse.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=argparse.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=argparse.RawDescriptionHelpFormatter,
+    )
+
+    logout_parser.set_defaults(func=CmdStudioToken)

dvc/commands/unprotect.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):

dvc/commands/update.py

@@ -1,12 +1,12 @@
 import argparse
-import logging
 
 from dvc.cli import completion
 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):

dvc/commands/version.py

@@ -1,11 +1,11 @@
 import argparse
-import logging
 
 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):

dvc/config.py

@@ -1,5 +1,4 @@
 """DVC config objects."""
-import logging
 import os
 import re
 from contextlib import contextmanager
@@ -9,14 +8,15 @@ from typing import TYPE_CHECKING, Dict, Optional
 from funcy import compact, memoize, re_find
 
 from dvc.exceptions import DvcException, NotDvcRepoError
+from dvc.log import logger
 
 from .utils.objects import cached_property
 
 if TYPE_CHECKING:
     from dvc.fs import FileSystem
-    from dvc.types import DictStrAny, StrPath
+    from dvc.types import DictStrAny
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class ConfigError(DvcException):
@@ -89,14 +89,14 @@ class Config(dict):
 
     def __init__(
         self,
-        dvc_dir: Optional["StrPath"] = None,
-        local_dvc_dir: Optional["StrPath"] = None,
+        dvc_dir: Optional[str] = None,
+        local_dvc_dir: Optional[str] = None,
         validate: bool = True,
         fs: Optional["FileSystem"] = None,
         config: Optional["DictStrAny"] = None,
         remote: Optional[str] = None,
         remote_config: Optional["DictStrAny"] = None,
-    ):  # pylint: disable=super-init-not-called
+    ):
         from dvc.fs import LocalFileSystem
 
         dvc_dir = os.fspath(dvc_dir) if dvc_dir else None
@@ -105,7 +105,7 @@ class Config(dict):
         self.fs = fs or self.wfs
 
         if dvc_dir:
-            self.dvc_dir = self.fs.path.abspath(dvc_dir)
+            self.dvc_dir = self.fs.abspath(dvc_dir)
 
         self.local_dvc_dir = local_dvc_dir
         if not fs and not local_dvc_dir:
@@ -145,10 +145,10 @@ class Config(dict):
         }
 
         if self.dvc_dir is not None:
-            files["repo"] = self.fs.path.join(self.dvc_dir, self.CONFIG)
+            files["repo"] = self.fs.join(self.dvc_dir, self.CONFIG)
 
         if self.local_dvc_dir is not None:
-            files["local"] = self.wfs.path.join(self.local_dvc_dir, self.CONFIG_LOCAL)
+            files["local"] = self.wfs.join(self.local_dvc_dir, self.CONFIG_LOCAL)
 
         return files
 
@@ -302,11 +302,11 @@ class Config(dict):
             return path.def_path
 
         if os.path.expanduser(path) != path:
-            return localfs.path.as_posix(path)
+            return localfs.as_posix(path)
 
         if isinstance(path, RelPath) or not os.path.isabs(path):
             path = relpath(path, conf_dir)
-            return localfs.path.as_posix(path)
+            return localfs.as_posix(path)
 
         return path
 
@@ -390,10 +390,10 @@ class Config(dict):
 
 
 def _parse_named(conf):
-    result: Dict[str, Dict] = {"remote": {}, "machine": {}}
+    result: Dict[str, Dict] = {"remote": {}, "machine": {}, "db": {}}
 
     for section, val in conf.items():
-        match = re_find(r'^\s*(remote|machine)\s*"(.*)"\s*$', section)
+        match = re_find(r'^\s*(remote|machine|db)\s*"(.*)"\s*$', section)
         if match:
             key, name = match
             result[key][name] = val
@@ -408,7 +408,7 @@ def _pack_named(conf):
     result = compact(conf)
 
     # Transform remote.name -> 'remote "name"'
-    for key in ("remote", "machine"):
+    for key in ("remote", "machine", "db"):
         for name, val in conf[key].items():
             result[f'{key} "{name}"'] = val
         result.pop(key, None)

dvc/config_schema.py

@@ -1,4 +1,3 @@
-import logging
 import os
 from urllib.parse import urlparse
 
@@ -16,7 +15,9 @@ from voluptuous import (
     Schema,
 )
 
-logger = logging.getLogger(__name__)
+from dvc.log import logger
+
+logger = logger.getChild(__name__)
 
 Bool = All(
     Lower,
@@ -256,6 +257,9 @@ SCHEMA = {
                     "exclude_shared_token_cache_credential": Bool,
                     "exclude_managed_identity_credential": Bool,
                     Optional("verify", default=False): Bool,
+                    "timeout": int,
+                    "read_timeout": int,
+                    "connection_timeout": int,
                     **REMOTE_COMMON,
                 },
                 "oss": {
@@ -314,6 +318,8 @@ SCHEMA = {
     "feature": FeatureSchema(
         {
             Optional("machine", default=False): Bool,
+            "dbt_profile": str,
+            "dbt_target": str,
         },
     ),
     "plots": {
@@ -346,4 +352,11 @@ SCHEMA = {
         "repo_url": str,
         Optional("offline", default=False): Bool,
     },
+    "db": {
+        str: {
+            "url": str,
+            "username": str,
+            "password": str,
+        },
+    },
 }

dvc/daemon.py

@@ -3,138 +3,185 @@
 import inspect
 import logging
 import os
-import platform
+import subprocess
 import sys
-from subprocess import Popen  # nosec B404
-from typing import List
+from contextlib import nullcontext
+from typing import TYPE_CHECKING, Any, List, Mapping, Optional, Sequence, Union
 
-from dvc.env import DVC_DAEMON
+from dvc.log import logger
+
+if TYPE_CHECKING:
+    from typing import ContextManager
+
+from dvc.env import DVC_DAEMON, DVC_DAEMON_LOGFILE
 from dvc.utils import fix_env, is_binary
+from dvc.utils.collections import ensure_list
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
-def _suppress_resource_warning(popen: Popen):
+def _suppress_resource_warning(popen: subprocess.Popen) -> None:
     """Sets the returncode to avoid ResourceWarning when popen is garbage collected."""
     # only use for daemon processes.
     # See https://bugs.python.org/issue38890.
     popen.returncode = 0
 
 
-def _popen(cmd, **kwargs) -> Popen:
-    prefix = [sys.executable]
-    if not is_binary():
-        main_entrypoint = os.path.join(
-            os.path.abspath(os.path.dirname(__file__)), "__main__.py"
-        )
-        prefix += [main_entrypoint]
-    return Popen(
-        prefix + cmd, close_fds=True, shell=False, **kwargs  # nosec B603  # noqa: S603
+def _win_detached_subprocess(args: Sequence[str], **kwargs) -> int:
+    assert os.name == "nt"
+
+    from subprocess import (  # type: ignore[attr-defined]
+        CREATE_NEW_PROCESS_GROUP,
+        CREATE_NO_WINDOW,
+        STARTF_USESHOWWINDOW,
+        STARTUPINFO,
     )
 
+    # https://stackoverflow.com/a/7006424
+    # https://bugs.python.org/issue41619
+    creationflags = CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW
 
-def _spawn_windows(cmd, env):
-    if sys.platform == "win32":
-        from subprocess import (  # nosec B404
-            CREATE_NEW_PROCESS_GROUP,
-            CREATE_NO_WINDOW,
-            STARTF_USESHOWWINDOW,
-            STARTUPINFO,
-        )
-
-        # https://stackoverflow.com/a/7006424
-        # https://bugs.python.org/issue41619
-        creationflags = CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW
-
-        startupinfo = STARTUPINFO()
-        startupinfo.dwFlags |= STARTF_USESHOWWINDOW
-
-        popen = _popen(
-            cmd, env=env, creationflags=creationflags, startupinfo=startupinfo
-        )
-        _suppress_resource_warning(popen)
+    startupinfo = STARTUPINFO()
+    startupinfo.dwFlags |= STARTF_USESHOWWINDOW
+    popen = subprocess.Popen(
+        args,
+        close_fds=True,
+        shell=False,  # noqa: S603
+        startupinfo=startupinfo,
+        creationflags=creationflags,
+        **kwargs,
+    )
+    _suppress_resource_warning(popen)
+    return popen.pid
 
 
-def _spawn_posix(cmd, env):
-    from dvc.cli import main
+def _get_dvc_args() -> List[str]:
+    args = [sys.executable]
+    if not is_binary():
+        root_dir = os.path.abspath(os.path.dirname(__file__))
+        main_entrypoint = os.path.join(root_dir, "__main__.py")
+        args.append(main_entrypoint)
+    return args
 
-    # `fork` will copy buffers, so we need to flush them before forking.
-    # Otherwise, we will get duplicated outputs.
-    if sys.stdout and not sys.stdout.closed:
-        sys.stdout.flush()
-    if sys.stderr and not sys.stderr.closed:
-        sys.stderr.flush()
+
+def _fork_process() -> int:
+    assert os.name == "posix"
 
     # NOTE: using os._exit instead of sys.exit, because dvc built
     # with PyInstaller has trouble with SystemExit exception and throws
     # errors such as "[26338] Failed to execute script __main__"
     try:
-        # pylint: disable-next=no-member
         pid = os.fork()  # type: ignore[attr-defined]
         if pid > 0:
-            return
+            return pid
     except OSError:
         logger.exception("failed at first fork")
-        os._exit(1)  # pylint: disable=protected-access
+        os._exit(1)
 
-    os.setsid()  # type: ignore[attr-defined]  # pylint: disable=no-member
+    os.setsid()  # type: ignore[attr-defined]
 
     try:
-        # pylint: disable-next=no-member
         pid = os.fork()  # type: ignore[attr-defined]
         if pid > 0:
-            os._exit(0)  # pylint: disable=protected-access
+            os._exit(0)
     except OSError:
         logger.exception("failed at second fork")
-        os._exit(1)  # pylint: disable=protected-access
+        os._exit(1)
 
-    sys.stdin.close()
-    sys.stdout.close()
-    sys.stderr.close()
-    os.closerange(0, 3)
-
-    if platform.system() == "Darwin":
-        # workaround for MacOS bug
-        # https://github.com/iterative/dvc/issues/4294
-        _popen(cmd, env=env).communicate()
-    else:
-        os.environ.update(env)
-        main(cmd)
-
-    os._exit(0)  # pylint: disable=protected-access
+    # disconnect from the terminal
+    fd = os.open(os.devnull, os.O_RDWR)
+    for fd2 in range(3):
+        os.dup2(fd, fd2)
+    os.close(fd)
+    return pid
 
 
-def _spawn(cmd, env):
-    logger.debug("Trying to spawn '%s'", cmd)
+def _posix_detached_subprocess(args: Sequence[str], **kwargs) -> int:
+    # double fork and execute a subprocess so that there are no zombies
+    read_end, write_end = os.pipe()
+    pid = _fork_process()
+    if pid > 0:  # in parent
+        os.close(write_end)
+        pid_str = os.read(read_end, 32).decode("utf8")
+        os.close(read_end)
+        return int(pid_str)
+
+    proc = subprocess.Popen(
+        args,
+        shell=False,  # noqa: S603
+        close_fds=True,
+        **kwargs,
+    )
+    os.close(read_end)
+    os.write(write_end, str(proc.pid).encode("utf8"))
+    os.close(write_end)
+
+    exit_code = proc.wait()
+    os._exit(exit_code)
+
+
+def _detached_subprocess(args: Sequence[str], **kwargs) -> int:
+    """Run in a detached subprocess."""
+    kwargs.setdefault("stdin", subprocess.DEVNULL)
+    kwargs.setdefault("stdout", subprocess.DEVNULL)
+    kwargs.setdefault("stderr", subprocess.DEVNULL)
 
     if os.name == "nt":
-        _spawn_windows(cmd, env)
-    elif os.name == "posix":
-        _spawn_posix(cmd, env)
-    else:
-        raise NotImplementedError
-
-    logger.debug("Spawned '%s'", cmd)
+        return _win_detached_subprocess(args, **kwargs)
+    return _posix_detached_subprocess(args, **kwargs)
 
 
-def daemon(args):
+def _map_log_level_to_flag() -> Optional[str]:
+    flags = {logging.DEBUG: "-v", logging.TRACE: "-vv"}  # type: ignore[attr-defined]
+    return flags.get(logger.getEffectiveLevel())
+
+
+def daemon(args: List[str]) -> None:
     """Launch a `dvc daemon` command in a detached process.
 
     Args:
         args (list): list of arguments to append to `dvc daemon` command.
     """
-    daemonize(["daemon", "-q", *args])
+    if flag := _map_log_level_to_flag():
+        args = [*args, flag]
+    daemonize(["daemon", *args])
 
 
-def daemonize(cmd: List[str]):
+def _spawn(
+    args: List[str],
+    executable: Optional[Union[str, List[str]]] = None,
+    env: Optional[Mapping[str, str]] = None,
+    output_file: Optional[str] = None,
+) -> int:
+    file: "ContextManager[Any]" = nullcontext()
+    kwargs = {}
+    if output_file:
+        file = open(output_file, "ab")  # noqa: SIM115
+        kwargs = {"stdout": file, "stderr": file}
+
+    if executable is None:
+        executable = _get_dvc_args()
+    else:
+        executable = ensure_list(executable)
+
+    with file:
+        return _detached_subprocess(executable + args, env=env, **kwargs)
+
+
+def daemonize(args: List[str], executable: Union[None, str, List[str]] = None) -> None:
+    if os.name not in ("posix", "nt"):
+        return None
+
     if os.environ.get(DVC_DAEMON):
         logger.debug("skipping launching a new daemon.")
         return
 
     env = fix_env()
+    env[DVC_DAEMON] = "1"
     if not is_binary():
         file_path = os.path.abspath(inspect.stack()[0][1])
         env["PYTHONPATH"] = os.path.dirname(os.path.dirname(file_path))
-    env[DVC_DAEMON] = "1"
 
-    _spawn(cmd, env)
+    logger.debug("Trying to spawn %r", args)
+    pid = _spawn(args, executable, env, output_file=env.get(DVC_DAEMON_LOGFILE))
+    logger.debug("Spawned %r with pid %s", args, pid)

dvc/dagascii.py

@@ -1,6 +1,5 @@
 """Draws DAG in ASCII."""
 
-import logging
 import math
 import os
 
@@ -8,7 +7,9 @@ from grandalf.graphs import Edge, Graph, Vertex
 from grandalf.layouts import SugiyamaLayout
 from grandalf.routing import EdgeViewer, route_with_lines
 
-logger = logging.getLogger(__name__)
+from dvc.log import logger
+
+logger = logger.getChild(__name__)
 
 
 class VertexViewer:
@@ -22,17 +23,16 @@ class VertexViewer:
     HEIGHT = 3  # top and bottom box edges + text
 
     def __init__(self, name):
-        # pylint: disable=invalid-name
         self._h = self.HEIGHT  # top and bottom box edges + text
         self._w = len(name) + 2  # right and left bottom edges + text
 
     @property
-    def h(self):  # pylint: disable=invalid-name
+    def h(self):
         """Height of the box."""
         return self._h
 
     @property
-    def w(self):  # pylint: disable=invalid-name
+    def w(self):
         """Width of the box."""
         return self._w
 
@@ -90,7 +90,6 @@ class AsciiCanvas:
             y1 (int): y coordinate where the line should end.
             char (str): character to draw the line with.
         """
-        # pylint: disable=too-many-arguments, too-many-branches
         if x0 > x1:
             x1, x0 = x0, x1
             y1, y0 = y0, y1
@@ -234,11 +233,11 @@ def draw(vertices, edges):
              | 1 |
              +---+
     """
-    # pylint: disable=too-many-locals
+
     # NOTE: coordinates might me negative, so we need to shift
     # everything to the positive plane before we actually draw it.
-    Xs = []  # noqa: N806, pylint: disable=invalid-name
-    Ys = []  # noqa: N806, pylint: disable=invalid-name
+    Xs = []  # noqa: N806
+    Ys = []  # noqa: N806
 
     sug = _build_sugiyama_layout(vertices, edges)
 
@@ -250,7 +249,7 @@ def draw(vertices, edges):
         Ys.append(vertex.view.xy[1] + vertex.view.h)
 
     for edge in sug.g.sE:
-        for x, y in edge.view._pts:  # pylint: disable=protected-access
+        for x, y in edge.view._pts:
             Xs.append(x)
             Ys.append(y)
 
@@ -266,7 +265,6 @@ def draw(vertices, edges):
 
     # NOTE: first draw edges so that node boxes could overwrite them
     for edge in sug.g.sE:
-        # pylint: disable=protected-access
         assert len(edge.view._pts) > 1
         for index in range(1, len(edge.view._pts)):
             start = edge.view._pts[index - 1]

dvc/data_cloud.py

@@ -1,10 +1,9 @@
 """Manages dvc remotes that user can use with push/pull/status commands."""
 
-import logging
 from typing import TYPE_CHECKING, Iterable, Optional, Set, Tuple
 
 from dvc.config import NoRemoteError, RemoteConfigError
-from dvc.fs.callbacks import Callback
+from dvc.log import logger
 from dvc.utils.objects import cached_property
 from dvc_data.hashfile.db import get_index
 from dvc_data.hashfile.transfer import TransferResult
@@ -15,7 +14,7 @@ if TYPE_CHECKING:
     from dvc_data.hashfile.hash_info import HashInfo
     from dvc_data.hashfile.status import CompareStatusResult
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class Remote:
@@ -36,11 +35,9 @@ class Remote:
 
         path = self.path
         if self.worktree:
-            path = self.fs.path.join(
-                path, ".dvc", CacheManager.FILES_DIR, DEFAULT_ALGORITHM
-            )
+            path = self.fs.join(path, ".dvc", CacheManager.FILES_DIR, DEFAULT_ALGORITHM)
         else:
-            path = self.fs.path.join(path, CacheManager.FILES_DIR, DEFAULT_ALGORITHM)
+            path = self.fs.join(path, CacheManager.FILES_DIR, DEFAULT_ALGORITHM)
         return get_odb(self.fs, path, hash_name=DEFAULT_ALGORITHM, **self.config)
 
     @cached_property
@@ -135,7 +132,7 @@ class DataCloud:
 
         remote = self.get_remote(name=name, command=command)
         if remote.fs.version_aware or remote.worktree:
-            raise NoRemoteError(
+            raise RemoteConfigError(
                 f"'{command}' is unsupported for cloud versioned remotes"
             )
         if hash_name in LEGACY_HASH_NAMES:
@@ -206,11 +203,13 @@ class DataCloud:
         jobs: Optional[int] = None,
         odb: "HashFileDB",
     ) -> "TransferResult":
+        from dvc.fs.callbacks import TqdmCallback
+
         if odb.hash_name == "md5-dos2unix":
             cache = self.repo.cache.legacy
         else:
             cache = self.repo.cache.local
-        with Callback.as_tqdm_callback(
+        with TqdmCallback(
             desc=f"Pushing to {odb.fs.unstrip_protocol(odb.path)}",
             unit="file",
         ) as cb:
@@ -265,11 +264,13 @@ class DataCloud:
         jobs: Optional[int] = None,
         odb: "HashFileDB",
     ) -> "TransferResult":
+        from dvc.fs.callbacks import TqdmCallback
+
         if odb.hash_name == "md5-dos2unix":
             cache = self.repo.cache.legacy
         else:
             cache = self.repo.cache.local
-        with Callback.as_tqdm_callback(
+        with TqdmCallback(
             desc=f"Fetching from {odb.fs.unstrip_protocol(odb.path)}",
             unit="file",
         ) as cb:

dvc/database/__init__.py

@@ -0,0 +1,22 @@
+from typing import Any, ContextManager, Optional, Union
+
+from . import dbt_query, sqla
+from .dbt_models import get_model
+from .serializer import export
+
+Client = Union[sqla.SQLAlchemyClient, dbt_query.DbtClient]
+
+
+def get_client(
+    config,
+    project_dir: Optional[str] = None,
+    profile: Optional[str] = None,
+    target: Optional[str] = None,
+    **kwargs: Any,
+) -> "ContextManager[Client]":
+    if config:
+        return sqla.client(config, **kwargs)
+    return dbt_query.client(project_dir=project_dir, profile=profile, target=target)
+
+
+__all__ = ["export", "get_client", "get_model", "Client"]

dvc/database/dbt_models.py

@@ -0,0 +1,40 @@
+from typing import List, Optional
+
+from dvc.log import logger
+
+from .dbt_runner import dbt_show
+from .dbt_utils import check_dbt
+from .serializer import AgateSerializer
+
+logger = logger.getChild(__name__)
+
+
+def ref(
+    name: str,
+    package: Optional[str] = None,
+    version: Optional[int] = None,
+) -> str:
+    parts: List[str] = []
+    if package:
+        parts.append(repr(package))
+
+    parts.append(repr(name))
+    if version:
+        parts.append(f"{version=}")
+
+    inner = ",".join(parts)
+    return "{{ ref(" + inner + ") }}"
+
+
+@check_dbt("model")
+def get_model(
+    name: str,
+    package: Optional[str] = None,
+    version: Optional[int] = None,
+    profile: Optional[str] = None,
+    target: Optional[str] = None,
+) -> AgateSerializer:
+    model = ref(name, package, version=version)
+    q = f"select * from {model}"  # noqa: S608
+    table = dbt_show(inline=q, profile=profile, target=target)
+    return AgateSerializer(table)

dvc/database/dbt_query.py

@@ -0,0 +1,88 @@
+import os
+from contextlib import contextmanager
+from typing import TYPE_CHECKING, Any, Callable, Dict, Iterator, Optional, Union
+
+from attrs import define, field
+
+from dvc.exceptions import DvcException
+from dvc.log import logger
+
+from .dbt_utils import check_dbt
+from .serializer import AgateSerializer, PandasSQLSerializer
+
+if TYPE_CHECKING:
+    from dbt.adapters.base.impl import BaseAdapter
+
+logger = logger.getChild(__name__)
+
+
+@define
+class DbtClient:
+    adapter: "BaseAdapter" = field(repr=lambda o: type(o).__qualname__)
+    creds: Dict[str, Any] = field(repr=False)
+
+    @contextmanager
+    def query(self, sql: str) -> Iterator[Union[AgateSerializer, PandasSQLSerializer]]:
+        from dbt.adapters.sql import SQLAdapter
+
+        with self.adapter.connection_named("execute"):
+            if isinstance(self.adapter, SQLAdapter):
+                # try to use connection to stream results
+                # otherwise fallback to fetch all at once
+                try:
+                    connection = self.adapter.connections.get_thread_connection().handle
+                except Exception:  # noqa: BLE001
+                    logger.debug("", exc_info=True)
+                else:
+                    # pandas raises warning that it does not support
+                    # dbapi2-based raw connection, but it does seem to work.
+                    yield PandasSQLSerializer(sql, connection)
+                    return
+            _, table = self.adapter.execute(sql, fetch=True)
+            yield AgateSerializer(table)
+
+    def test_connection(self, onerror: Optional[Callable[[], Any]] = None) -> None:
+        with self.adapter.connection_named("debug"):
+            try:
+                self.adapter.debug_query()
+            except Exception as exc:
+                if callable(onerror):
+                    onerror()
+
+                logger.exception(
+                    "dvc was unable to connect to the specified database. "
+                    "Please check your database credentials and try again.",
+                    exc_info=False,
+                )
+                raise DvcException("The database returned the following error") from exc
+
+
+@contextmanager
+@check_dbt("query")
+def client(
+    project_dir: Optional[str] = None,
+    profiles_dir: Optional[str] = None,
+    profile: Optional[str] = None,
+    target: Optional[str] = None,
+) -> Iterator["DbtClient"]:
+    from dbt.adapters import factory as adapters_factory
+    from dbt.adapters.base.impl import BaseAdapter
+
+    from .dbt_utils import get_or_build_profile, get_profiles_dir, init_dbt
+
+    profiles_dir = profiles_dir or get_profiles_dir(project_dir)
+    flags = init_dbt(profiles_dir, os.getcwd(), target=target)
+
+    with flags, adapters_factory.adapter_management():
+        # likely invalid connection profile or no adapter
+        profile_obj = get_or_build_profile(project_dir, profile, target)
+
+        adapters_factory.register_adapter(profile_obj)  # type: ignore[arg-type]
+        adapter = adapters_factory.get_adapter(profile_obj)  # type: ignore[arg-type]
+        assert isinstance(adapter, BaseAdapter)
+        try:
+            creds = dict(profile_obj.credentials.connection_info())
+        except:  # noqa: E722
+            creds = {}
+
+        yield DbtClient(adapter, creds)

dvc/database/dbt_runner.py

@@ -0,0 +1,61 @@
+from typing import TYPE_CHECKING, List, Optional, Sequence, Union
+
+from dvc.exceptions import DvcException
+from dvc.log import logger
+
+if TYPE_CHECKING:
+    from agate import Table
+    from dbt.contracts.results import RunResult
+
+logger = logger.getChild(__name__)
+
+
+class DbtInternalError(DvcException):
+    pass
+
+
+def _kw_to_cmd_args(**kwargs: Union[None, bool, str, int]) -> List[str]:
+    args: List[str] = []
+    for key, value in kwargs.items():
+        key = key.replace("_", "-")
+        if value is None:
+            continue  # skip creating a flag in this case
+        if value is True:
+            args.append(f"--{key}")
+        elif value is False:
+            args.append(f"--no-{key}")
+        else:
+            args.extend([f"--{key}", str(value)])
+    return args
+
+
+def _dbt_invoke(*posargs: str, quiet: bool = True, **kw: Union[None, bool, str, int]):
+    from dbt.cli.main import dbtRunner
+
+    args = _kw_to_cmd_args(quiet=quiet or None)  # global options
+    args.extend([*posargs, *_kw_to_cmd_args(**kw)])
+
+    runner = dbtRunner()
+    result = runner.invoke(args)
+    if result.success:
+        return result.result
+    raise DbtInternalError(f"failed to run dbt {posargs[0]}") from result.exception
+
+
+def dbt_show(
+    inline: Optional[str] = None,
+    limit: int = -1,
+    profile: Optional[str] = None,
+    target: Optional[str] = None,
+) -> "Table":
+    from dbt.contracts.results import RunExecutionResult
+
+    result = _dbt_invoke(
+        "show", inline=inline, limit=limit, profile=profile, target=target
+    )
+    assert isinstance(result, RunExecutionResult)
+
+    run_results: Sequence["RunResult"] = result.results
+    run_result, *_ = run_results
+    assert run_result.agate_table is not None
+    return run_result.agate_table

dvc/database/dbt_utils.py

@@ -0,0 +1,100 @@
+import os
+from contextlib import contextmanager
+from importlib.util import find_spec
+from typing import TYPE_CHECKING, Optional
+
+from funcy import cut_prefix
+
+from dvc.exceptions import DvcException
+from dvc.log import logger
+from dvc.utils import packaging
+
+if TYPE_CHECKING:
+    from dbt.config.profile import Profile
+
+logger = logger.getChild(__name__)
+
+DBT_PROJECT_FILE = "dbt_project.yml"
+
+
+@contextmanager
+def check_dbt(action: Optional[str] = None):
+    if not (find_spec("dbt") and find_spec("dbt.cli")):
+        action = f" {action}" if action else ""
+        raise DvcException(f"Could not run{action}. dbt-core is not installed")
+    yield packaging.check_required_version(pkg="dbt-core")
+
+
+@contextmanager
+def init_dbt(
+    profiles_dir: str,
+    project_dir: str,
+    target: Optional[str] = None,
+):
+    from argparse import Namespace
+
+    from dbt import flags
+
+    prev = flags.get_flags()
+    try:
+        args = Namespace(
+            use_colors=True,
+            project_dir=project_dir,
+            profiles_dir=profiles_dir,
+            target=target,
+        )
+        flags.set_from_args(args, None)
+        yield flags.get_flags()
+    finally:
+        flags.set_flags(prev)
+
+
+def get_profiles_dir(project_dir: Optional[str] = None) -> str:
+    from dbt.cli.resolvers import default_profiles_dir
+
+    if profiles_dir := os.getenv("DBT_PROFILES_DIR"):
+        return profiles_dir
+    if project_dir and os.path.isfile(os.path.join(project_dir, "profiles.yml")):
+        return project_dir
+    return os.fspath(default_profiles_dir())
+
+
+@contextmanager
+def _handle_profile_parsing_error():
+    from dbt.exceptions import DbtRuntimeError
+
+    try:
+        yield
+    except DbtRuntimeError as e:
+        cause = e.__cause__ is not None and e.__cause__.__context__
+        if isinstance(cause, ModuleNotFoundError) and (
+            adapter := cut_prefix(cause.name, "dbt.adapters.")
+        ):
+            # DbtRuntimeError is very noisy, so send it to debug
+            logger.debug("", exc_info=True)
+            raise DvcException(f"dbt-{adapter} dependency is missing") from cause
+        raise DvcException("failed to read connection profiles") from e
+
+
+def is_dbt_project(project_dir: str):
+    return os.path.isfile(os.path.join(project_dir, DBT_PROJECT_FILE))
+
+
+@_handle_profile_parsing_error()
+def get_or_build_profile(
+    project_dir: Optional[str], profile: Optional[str], target: Optional[str]
+) -> "Profile":
+    from dbt.config.profile import Profile
+    from dbt.config.renderer import ProfileRenderer
+    from dbt.config.runtime import load_profile
+
+    if project_dir and is_dbt_project(project_dir):
+        return load_profile(
+            project_dir, {}, profile_name_override=profile, target_override=target
+        )
+
+    if not profile:
+        raise DvcException("No profile specified to query from.")
+
+    renderer = ProfileRenderer({})
+    return Profile.render(renderer, profile, target_override=target)

dvc/database/serializer.py

@@ -0,0 +1,54 @@
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Union
+
+if TYPE_CHECKING:
+    import sqlalchemy as sa
+    from agate import Table
+
+
+def noop(_):
+    pass
+
+
+@dataclass
+class PandasSQLSerializer:
+    sql: "Union[sa.TextClause, str]"
+    con: "sa.Connection"
+    chunksize: int = 10_000
+
+    def to_csv(self, file: str, progress=noop) -> None:
+        import pandas as pd
+
+        with open(file, mode="wb") as f:
+            idfs = pd.read_sql_query(self.sql, self.con, chunksize=self.chunksize)
+            for i, df in enumerate(idfs):
+                df.to_csv(f, header=i == 0, index=False)
+                progress(len(df))
+
+    def to_json(self, file: str, progress=noop) -> None:  # noqa: ARG002
+        import pandas as pd
+
+        df = pd.read_sql_query(self.sql, self.con)
+        df.to_json(file, orient="records")
+
+
+@dataclass
+class AgateSerializer:
+    table: "Table"
+
+    def to_csv(self, file: str, progress=noop) -> None:  # noqa: ARG002
+        return self.table.to_csv(file)
+
+    def to_json(self, file: str, progress=noop) -> None:  # noqa: ARG002
+        return self.table.to_json(file)
+
+
+def export(
+    serializer: Union[PandasSQLSerializer, AgateSerializer],
+    file: str,
+    format: str = "csv",  # noqa: A002
+    progress=noop,
+) -> None:
+    if format == "csv":
+        return serializer.to_csv(file, progress=progress)
+    return serializer.to_json(file, progress=progress)

dvc/database/sqla.py

@@ -0,0 +1,73 @@
+from contextlib import contextmanager
+from dataclasses import dataclass
+from typing import Any, Callable, Dict, Iterator, Optional, Union
+
+from sqlalchemy import create_engine, text
+from sqlalchemy.engine import URL, Engine
+from sqlalchemy.engine import make_url as _make_url
+from sqlalchemy.exc import NoSuchModuleError
+
+from dvc.exceptions import DvcException
+from dvc.log import logger
+
+from .serializer import PandasSQLSerializer
+
+logger = logger.getChild(__name__)
+
+
+def make_url(url: Union[URL, str], **kwargs: Any) -> URL:
+    return _make_url(url).set(**kwargs)
+
+
+def url_from_config(config: Union[str, URL, Dict[str, str]]) -> URL:
+    if isinstance(config, (str, URL)):
+        return make_url(config)
+    return make_url(**config)
+
+
+@dataclass
+class SQLAlchemyClient:
+    engine: Engine
+
+    @contextmanager
+    def query(self, sql: str) -> Iterator[PandasSQLSerializer]:
+        with self.engine.connect().execution_options(stream_results=True) as conn:
+            yield PandasSQLSerializer(text(sql), conn)
+
+    def test_connection(self, onerror: Optional[Callable[[], Any]] = None) -> None:
+        try:
+            with self.engine.connect() as conn:
+                conn.execute(text("select 1"))
+        except Exception as exc:
+            if callable(onerror):
+                onerror()
+            logger.exception(
+                "dvc was unable to connect to the specified database. "
+                "Please check your database credentials and try again.",
+                exc_info=False,
+            )
+            raise DvcException("The database returned the following error") from exc
+
+
+@contextmanager
+def handle_error(url: URL):
+    try:
+        yield
+    except (ModuleNotFoundError, NoSuchModuleError) as e:
+        # TODO: write installation instructions
+        raise DvcException(
+            f"Could not load database driver for {url.drivername!r}"
+        ) from e
+
+
+@contextmanager
+def client(
+    url_or_config: Union[str, URL, Dict[str, str]], **engine_kwargs: Any
+) -> Iterator[SQLAlchemyClient]:
+    url = url_from_config(url_or_config)
+    with handle_error(url):
+        engine = create_engine(url, **engine_kwargs)
+    try:
+        yield SQLAlchemyClient(engine)
+    finally:
+        engine.dispose()

dvc/dependency/__init__.py

@@ -1,9 +1,10 @@
 from collections import defaultdict
-from typing import Any, Mapping, Set
+from typing import Any, Dict, List, Mapping, Set
 
 from dvc.output import ARTIFACT_SCHEMA, DIR_FILES_SCHEMA, Output
 
 from .base import Dependency
+from .db import DB_SCHEMA, PARAM_DB, DbDependency, DbtDependency
 from .param import ParamsDependency
 from .repo import RepoDependency
 
@@ -14,20 +15,28 @@ from .repo import RepoDependency
 SCHEMA: Mapping[str, Any] = {
     **ARTIFACT_SCHEMA,
     **RepoDependency.REPO_SCHEMA,
+    **DB_SCHEMA,
     Output.PARAM_FILES: [DIR_FILES_SCHEMA],
     Output.PARAM_FS_CONFIG: dict,
 }
 
 
 def _get(stage, p, info, **kwargs):
-    if info and info.get(RepoDependency.PARAM_REPO):
-        repo = info.pop(RepoDependency.PARAM_REPO)
-        return RepoDependency(repo, stage, p, info)
+    d = info or {}
+    db = d.get(PARAM_DB, {})
+    params = d.pop(ParamsDependency.PARAM_PARAMS, None)
+    repo = d.pop(RepoDependency.PARAM_REPO, None)
 
-    if info and info.get(ParamsDependency.PARAM_PARAMS):
-        params = info.pop(ParamsDependency.PARAM_PARAMS)
+    if params:
         return ParamsDependency(stage, p, params)
+    if DbDependency.PARAM_QUERY in db:
+        return DbDependency(stage, info)
+    if db:
+        return DbtDependency(repo, stage, info)
 
+    assert p
+    if repo:
+        return RepoDependency(repo, stage, p, info)
     return Dependency(stage, p, info, **kwargs)
 
 
@@ -44,9 +53,11 @@ def loadd_from(stage, d_list):
     return ret
 
 
-def loads_from(stage, s_list, erepo=None, fs_config=None):
+def loads_from(stage, s_list, erepo=None, fs_config=None, db=None):
     assert isinstance(s_list, list)
     info = {RepoDependency.PARAM_REPO: erepo} if erepo else {}
+    if db:
+        info.update({"db": db})
     return [
         _get(
             stage,
@@ -58,7 +69,7 @@ def loads_from(stage, s_list, erepo=None, fs_config=None):
     ]
 
 
-def _merge_params(s_list):
+def _merge_params(s_list) -> Dict[str, List[str]]:
     d = defaultdict(list)
     default_file = ParamsDependency.DEFAULT_PARAMS_FILE
 

dvc/dependency/base.py

@@ -33,7 +33,7 @@ class Dependency(Output):
         if self.fs.version_aware:
             old_fs_path = self.fs_path
             try:
-                self.fs_path = self.fs.path.version_path(self.fs_path, None)
+                self.fs_path = self.fs.version_path(self.fs_path, None)
                 if self.changed_meta():
                     return {str(self): "update available"}
             finally:
@@ -42,9 +42,9 @@ class Dependency(Output):
 
     def update(self, rev=None):
         if self.fs.version_aware:
-            self.fs_path = self.fs.path.version_path(self.fs_path, rev)
+            self.fs_path = self.fs.version_path(self.fs_path, rev)
             self.meta = self.get_meta()
-            self.fs_path = self.fs.path.version_path(self.fs_path, self.meta.version_id)
+            self.fs_path = self.fs.version_path(self.fs_path, self.meta.version_id)
 
     def download(self, to, jobs=None):
         fs_download(self.fs, self.fs_path, to.fs_path, jobs=jobs)
@@ -52,7 +52,7 @@ class Dependency(Output):
     def save(self):
         super().save()
         if self.fs.version_aware:
-            self.fs_path = self.fs.path.version_path(self.fs_path, self.meta.version_id)
+            self.fs_path = self.fs.version_path(self.fs_path, self.meta.version_id)
 
     def dumpd(self, **kwargs):
         if self.fs.version_aware:

dvc/dependency/db.py

@@ -0,0 +1,324 @@
+import os
+from contextlib import contextmanager, nullcontext
+from typing import TYPE_CHECKING, Any, Callable, Dict, Iterator, Optional, Union
+
+from funcy import compact, log_durations
+
+from dvc.exceptions import DvcException
+from dvc.log import logger
+from dvc.scm import SCM
+
+from .base import Dependency
+
+if TYPE_CHECKING:
+    from rich.status import Status
+
+    from dvc.output import Output
+    from dvc.stage import Stage
+
+logger = logger.getChild(__name__)
+
+
+PARAM_DB = "db"
+PARAM_PROFILE = "profile"
+PARAM_FILE_FORMAT = "file_format"
+
+
+def _get_dbt_config(config: Dict) -> Dict:
+    conf = config.get("feature", {})
+    pref = "dbt_"
+    return {k.lstrip(pref): v for k, v in conf.items() if k.startswith(pref)}
+
+
+@contextmanager
+def log_status(
+    msg, status: Optional["Status"] = None, log=logger.debug
+) -> Iterator["Status"]:
+    from funcy import log_durations
+
+    from dvc.ui import ui
+
+    with log_durations(log, msg), status or ui.status(msg) as st:
+        st.update(msg)
+        yield st
+
+
+@contextmanager
+def chdir(path):
+    wdir = os.getcwd()
+    os.chdir(path)
+    try:
+        yield
+    finally:
+        os.chdir(wdir)
+
+
+@contextmanager
+def download_progress(to: "Output") -> Iterator[Callable[[int], Any]]:
+    from dvc.ui import ui
+    from dvc.ui._rich_progress import DbDownloadProgress
+
+    with log_durations(logger.debug, f"Saving to {to}"), DbDownloadProgress(
+        console=ui.error_console,
+    ) as progress:
+        task = progress.add_task("Saving", total=None, output=to)
+        yield lambda n: progress.advance(task, advance=n)
+        progress.update(task, description="Saved", total=0)
+
+
+class AbstractDependency(Dependency):
+    """Dependency without workspace/fs/fs_path"""
+
+    def __init__(self, stage: "Stage", info: Dict[str, Any], *args, **kwargs):
+        self.repo = stage.repo
+        self.stage = stage
+        self.fs = None
+        self.fs_path = None
+        self.def_path = None  # type: ignore[assignment]
+        self.info = info or {}
+
+
+class DbDependency(AbstractDependency):
+    PARAM_QUERY = "query"
+    PARAM_CONNECTION = "connection"
+    QUERY_SCHEMA = {PARAM_QUERY: str, PARAM_CONNECTION: str}
+
+    def __init__(self, stage: "Stage", info, *args, **kwargs):
+        super().__init__(stage, info, *args, **kwargs)
+        self.target = None
+
+    def __repr__(self):
+        return "{}:{}".format(
+            self.__class__.__name__, "".join(f"{k}=={v}" for k, v in self.info.items())
+        )
+
+    def __str__(self):
+        from dvc.utils.humanize import truncate_text
+
+        db_info = self.info.get(PARAM_DB, {})
+        query = db_info.get(self.PARAM_QUERY, "[query]")
+        return truncate_text(query, 50)
+
+    def workspace_status(self):
+        return False  # no workspace to check
+
+    def status(self):
+        return self.workspace_status()
+
+    def save(self):
+        """nothing to save."""
+
+    def dumpd(self, **kwargs):
+        db_info = compact(self.info.get(PARAM_DB, {}))
+        return {PARAM_DB: db_info} if db_info else {}
+
+    def update(self, rev=None):
+        """nothing to update."""
+
+    def download(
+        self,
+        to: "Output",
+        jobs: Optional[int] = None,  # noqa: ARG002
+        file_format: Optional[str] = None,
+        **kwargs: Any,
+    ) -> None:
+        from dvc.database import export, get_client
+
+        db_info = self.info.get(PARAM_DB, {})
+        query = db_info.get(self.PARAM_QUERY)
+        if not query:
+            raise DvcException("Cannot download: no query specified")
+
+        dbt_config = _get_dbt_config(self.repo.config)
+        profile = db_info.get(PARAM_PROFILE) or dbt_config.get(PARAM_PROFILE)
+        target = self.target or dbt_config.get("target")
+
+        connection = db_info.get(self.PARAM_CONNECTION)
+        db_config = self.repo.config.get("db", {})
+        config = db_config.get(connection)
+        if connection and not config:
+            raise DvcException(f"connection {connection} not found in config")
+
+        project_dir = self.repo.root_dir
+        with get_client(
+            config, project_dir=project_dir, profile=profile, target=target
+        ) as db:
+            logger.debug("using adapter: %s", db)
+            with log_status("Testing connection") as status:
+                db.test_connection(onerror=status.stop)
+
+            file_format = file_format or db_info.get(PARAM_FILE_FORMAT, "csv")
+            assert file_format
+            with log_status("Executing query") as status, db.query(query) as serializer:
+                status.stop()
+                logger.debug("using serializer: %s", serializer)
+                with download_progress(to) as progress:
+                    return export(
+                        serializer, to.fs_path, format=file_format, progress=progress
+                    )
+
+
+class DbtDependency(AbstractDependency):
+    PARAM_MODEL = "model"
+    PARAM_VERSION = "version"
+    PARAM_PROJECT_DIR = "project_dir"
+    DBT_SCHEMA = {
+        PARAM_MODEL: str,
+        PARAM_VERSION: str,
+        PARAM_PROJECT_DIR: str,
+    }
+
+    def __init__(self, def_repo: Dict[str, Any], stage: "Stage", info, *args, **kwargs):
+        self.def_repo = def_repo or {}
+        self.target = None
+        super().__init__(stage, info, *args, **kwargs)
+
+    def __repr__(self):
+        return "{}:{}".format(
+            self.__class__.__name__,
+            "".join(f"{k}=={v}" for k, v in {**self.def_repo, **self.info}.items()),
+        )
+
+    def __str__(self):
+        from .repo import RepoDependency
+
+        repo = self.def_repo.get(RepoDependency.PARAM_URL)
+        rev = self.def_repo.get(RepoDependency.PARAM_REV)
+
+        db_info = self.info.get(PARAM_DB, {})
+        db = db_info.get(self.PARAM_MODEL, "")
+        project_dir = db_info.get(self.PARAM_PROJECT_DIR, "")
+        repo_info = ""
+        if repo:
+            repo_info += repo
+        if rev:
+            repo_info += f"@{rev}"
+        if project_dir:
+            repo_info += f":/{project_dir}"
+        return db + (f"({repo_info})" if repo_info else "")
+
+    @property
+    def locked_rev(self):
+        from .repo import RepoDependency
+
+        return self.def_repo.get(RepoDependency.PARAM_REV_LOCK)
+
+    @property
+    def rev(self):
+        from .repo import RepoDependency
+
+        return self.def_repo.get(RepoDependency.PARAM_REV)
+
+    def workspace_status(self):
+        if not self.def_repo:
+            return
+
+        current = self._get_clone(self.locked_rev or self.rev).get_rev()
+        updated = self._get_clone(self.rev).get_rev()
+        if current != updated:
+            return {str(self): "update available"}
+        return {}
+
+    def status(self):
+        return self.workspace_status()
+
+    def save(self):
+        from .repo import RepoDependency
+
+        if not self.def_repo:
+            return
+
+        rev = self._get_clone(self.locked_rev or self.rev).get_rev()
+        if self.def_repo.get(RepoDependency.PARAM_REV_LOCK) is None:
+            self.def_repo[RepoDependency.PARAM_REV_LOCK] = rev
+
+    def dumpd(self, **kwargs) -> Dict[str, Union[str, Dict[str, str]]]:
+        from .repo import RepoDependency
+
+        def_repo = {}
+        if self.def_repo:
+            def_repo = RepoDependency._dump_def_repo(self.def_repo)
+
+        db_info = compact(self.info.get(PARAM_DB, {}))
+        return compact({RepoDependency.PARAM_REPO: def_repo, PARAM_DB: db_info})
+
+    def _get_clone(self, rev):
+        from dvc.repo.open_repo import _cached_clone
+
+        from .repo import RepoDependency
+
+        url = self.def_repo.get(RepoDependency.PARAM_URL)
+        repo_root = self.repo.root_dir if self.repo else os.getcwd()
+        return SCM(_cached_clone(url, rev) if url else repo_root)
+
+    def update(self, rev=None):
+        from .repo import RepoDependency
+
+        if not self.def_repo:
+            return
+
+        if rev:
+            self.def_repo[RepoDependency.PARAM_REV] = rev
+        else:
+            rev = self.rev
+        self.def_repo[RepoDependency.PARAM_REV_LOCK] = self._get_clone(rev).get_rev()
+
+    def download(
+        self,
+        to: "Output",
+        jobs: Optional[int] = None,  # noqa: ARG002
+        file_format: Optional[str] = None,
+    ) -> None:
+        from .repo import RepoDependency
+
+        project_dir = self.info.get(PARAM_DB, {}).get(self.PARAM_PROJECT_DIR, "")
+        if self.def_repo:
+            ctx = repo = self._get_clone(self.locked_rev or self.rev)
+            self.def_repo[RepoDependency.PARAM_REV_LOCK] = repo.get_rev()
+            root = wdir = repo.root_dir
+        else:
+            ctx = nullcontext()
+            root = self.repo.root_dir
+            wdir = self.stage.wdir
+
+        project_path = os.path.join(wdir, project_dir) if project_dir else root
+        with ctx, chdir(project_path):
+            self._download_db(to, file_format=file_format)
+
+    def _download_db(
+        self,
+        to: "Output",
+        version: Optional[int] = None,
+        file_format: Optional[str] = None,
+    ) -> None:
+        from dvc.database import export, get_model
+
+        db_info = self.info.get(PARAM_DB, {})
+        model = db_info.get(self.PARAM_MODEL)
+        version = version or db_info.get(self.PARAM_VERSION)
+        file_format = file_format or db_info.get(PARAM_FILE_FORMAT) or "csv"
+        assert file_format
+        if not model:
+            raise DvcException("Cannot download, no model specified")
+
+        dbt_config = _get_dbt_config(self.repo.config)
+        profile = db_info.get(PARAM_PROFILE) or dbt_config.get(PARAM_PROFILE)
+        target = self.target or db_info.get("target") or dbt_config.get("target")
+
+        with log_status("Downloading model"):
+            serializer = get_model(
+                model, version=version, profile=profile, target=target
+            )
+        # NOTE: we keep everything in memory, and then export it out later.
+        with download_progress(to) as progress:
+            export(serializer, to.fs_path, format=file_format, progress=progress)
+
+
+DB_SCHEMA = {
+    PARAM_DB: {
+        PARAM_PROFILE: str,
+        PARAM_FILE_FORMAT: str,
+        **DbDependency.QUERY_SCHEMA,
+        **DbtDependency.DBT_SCHEMA,
+    },
+}