deps: bump dvc-data to >=3.3

Leftover from https://github.com/iterative/dvc/pull/10190

.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

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

.github/workflows/codeql.yml

@@ -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.9.1
-    hooks:
-      - id: black
-  - repo: https://github.com/PyCQA/isort
-    rev: 5.12.0
-    hooks:
-      - id: isort
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.0.291'
+    rev: 'v0.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,8 +1,7 @@
 from dvc.fs.dvc import _DVCFileSystem as DVCFileSystem
 
 from .artifacts import artifacts_show
-from .data import open  # pylint: disable=redefined-builtin
-from .data import get_url, read
+from .data import get_url, open, read
 from .experiments import exp_save, exp_show
 from .scm import all_branches, all_commits, all_tags
 from .show import metrics_show, params_show

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/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)
@@ -86,7 +86,7 @@ class CacheManager:
         return FileStorage(
             key=(),
             fs=self.local.fs,
-            path=self.local.fs.path.join(self.default_local_cache_dir, self.FS_DIR),
+            path=self.local.fs.join(self.default_local_cache_dir, self.FS_DIR),
         )
 
     def _init_odb(self, schemes):
@@ -137,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,6 +1,5 @@
 """Main parser for the dvc cli."""
 import argparse
-import logging
 import os
 from functools import lru_cache
 
@@ -20,6 +19,7 @@ from dvc.commands import (
     data_sync,
     destroy,
     diff,
+    du,
     experiments,
     freeze,
     gc,
@@ -27,6 +27,7 @@ from dvc.commands import (
     get_url,
     git_hook,
     imp,
+    imp_db,
     imp_url,
     init,
     install,
@@ -42,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,
@@ -65,6 +68,7 @@ COMMANDS = [
     data_sync,
     gc,
     imp,
+    imp_db,
     imp_url,
     config,
     checkout,
@@ -91,16 +95,18 @@ COMMANDS = [
     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
@@ -112,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

@@ -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, fix_subparsers
 from dvc.exceptions import DvcException
+from dvc.log import logger
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdArtifactsGet(CmdBaseNoRepo):

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

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

@@ -1,16 +1,24 @@
 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):
     def run(self):
-        self.repo.experiments.apply(self.args.experiment, force=self.args.force)
+        if not self.args.force:
+            ui.write(
+                "The --no-force option is deprecated and will be removed in a future"
+                " DVC release. To revert the result of 'exp apply', run:\n"
+                "\n\tgit reset --hard\n"
+                "\tgit stash apply refs/exps/apply/stash\n"
+            )
+        self.repo.experiments.apply(self.args.experiment)
 
         return 0
 
@@ -25,11 +33,10 @@ def add_parser(experiments_subparsers, parent_parser):
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )
     experiments_apply_parser.add_argument(
-        "-f",
-        "--force",
-        action="store_true",
-        default=False,
-        help="Do not prompt when removing working directory files.",
+        "--no-force",
+        action="store_false",
+        dest="force",
+        help="Fail if this command would overwrite conflicting changes.",
     )
     experiments_apply_parser.add_argument(
         "experiment", help="Experiment to be applied."

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

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

@@ -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 CmdExperimentsRename(CmdBase):

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

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

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
 
@@ -82,6 +86,7 @@ class CmdGC(CmdBase):
             num=self.args.num,
             not_in_remote=self.args.not_in_remote,
             dry=self.args.dry,
+            skip_failed=self.args.skip_failed,
         )
         return 0
 
@@ -184,6 +189,12 @@ def add_parser(subparsers, parent_parser):
         help="Remote storage to collect garbage in",
         metavar="<name>",
     )
+    gc_parser.add_argument(
+        "--skip-failed",
+        action="store_true",
+        default=False,
+        help="Skip revisions that fail when collected.",
+    )
     gc_parser.add_argument(
         "-f",
         "--force",

dvc/commands/get.py

@@ -1,12 +1,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,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 CmdGetUrl(CmdBaseNoRepo):

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

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

dvc/commands/metrics.py

@@ -1,13 +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.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

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,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, fix_subparsers
+from dvc.log import logger
 from dvc.ui import ui
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class CmdParamsDiff(CmdBase):

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,132 +3,185 @@
 import inspect
 import logging
 import os
+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)
-
-    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
@@ -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

@@ -4,6 +4,7 @@ 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,

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,
+    },
+}

dvc/dependency/param.py

@@ -1,4 +1,3 @@
-import logging
 import os
 import typing
 from collections import defaultdict
@@ -7,6 +6,7 @@ from typing import TYPE_CHECKING, Any, Dict, List, Optional
 import dpath
 
 from dvc.exceptions import DvcException
+from dvc.log import logger
 from dvc.utils.serialize import ParseError, load_path
 from dvc_data.hashfile.hash_info import HashInfo
 
@@ -15,7 +15,7 @@ from .base import Dependency
 if TYPE_CHECKING:
     from dvc.fs import FileSystem
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class MissingParamsError(DvcException):
@@ -54,12 +54,15 @@ def read_param_file(
                 continue
         return ret
 
-    from dpath import merge
+    from copy import deepcopy
 
-    for key_path in key_paths:
+    from dpath import merge
+    from funcy import distinct
+
+    for key_path in distinct(key_paths):
         merge(
             ret,
-            dpath.search(config, key_path, separator="."),
+            deepcopy(dpath.search(config, key_path, separator=".")),
             separator=".",
         )
     return ret
@@ -132,8 +135,8 @@ class ParamsDependency(Dependency):
         from funcy import ldistinct
 
         status: Dict[str, Any] = defaultdict(dict)
-        assert isinstance(self.hash_info.value, dict)
         info = self.hash_info.value if self.hash_info else {}
+        assert isinstance(info, dict)
         actual = self.read_params()
 
         # NOTE: we want to preserve the order of params as specified in the

dvc/dependency/repo.py

@@ -1,7 +1,7 @@
 from copy import deepcopy
-from typing import TYPE_CHECKING, Dict, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, Optional, Union
 
-from voluptuous import Any, Required
+import voluptuous as vol
 
 from dvc.utils import as_posix
 
@@ -22,11 +22,11 @@ class RepoDependency(Dependency):
 
     REPO_SCHEMA = {
         PARAM_REPO: {
-            Required(PARAM_URL): str,
+            vol.Required(PARAM_URL): str,
             PARAM_REV: str,
             PARAM_REV_LOCK: str,
-            PARAM_CONFIG: Any(str, dict),
-            PARAM_REMOTE: Any(str, dict),
+            PARAM_CONFIG: vol.Any(str, dict),
+            PARAM_REMOTE: vol.Any(str, dict),
         }
     }
 
@@ -64,28 +64,31 @@ class RepoDependency(Dependency):
         if self.def_repo.get(self.PARAM_REV_LOCK) is None:
             self.def_repo[self.PARAM_REV_LOCK] = rev
 
-    def dumpd(self, **kwargs) -> Dict[str, Union[str, Dict[str, str]]]:
-        repo = {self.PARAM_URL: self.def_repo[self.PARAM_URL]}
+    @classmethod
+    def _dump_def_repo(cls, def_repo) -> Dict[str, str]:
+        repo = {cls.PARAM_URL: def_repo[cls.PARAM_URL]}
 
-        rev = self.def_repo.get(self.PARAM_REV)
+        rev = def_repo.get(cls.PARAM_REV)
         if rev:
-            repo[self.PARAM_REV] = self.def_repo[self.PARAM_REV]
+            repo[cls.PARAM_REV] = def_repo[cls.PARAM_REV]
 
-        rev_lock = self.def_repo.get(self.PARAM_REV_LOCK)
+        rev_lock = def_repo.get(cls.PARAM_REV_LOCK)
         if rev_lock:
-            repo[self.PARAM_REV_LOCK] = rev_lock
+            repo[cls.PARAM_REV_LOCK] = rev_lock
 
-        config = self.def_repo.get(self.PARAM_CONFIG)
+        config = def_repo.get(cls.PARAM_CONFIG)
         if config:
-            repo[self.PARAM_CONFIG] = config
+            repo[cls.PARAM_CONFIG] = config
 
-        remote = self.def_repo.get(self.PARAM_REMOTE)
+        remote = def_repo.get(cls.PARAM_REMOTE)
         if remote:
-            repo[self.PARAM_REMOTE] = remote
+            repo[cls.PARAM_REMOTE] = remote
+        return repo
 
+    def dumpd(self, **kwargs) -> Dict[str, Union[str, Dict[str, str]]]:
         return {
             self.PARAM_PATH: self.def_path,
-            self.PARAM_REPO: repo,
+            self.PARAM_REPO: self._dump_def_repo(self.def_repo),
         }
 
     def update(self, rev: Optional[str] = None):

dvc/dirs.py

@@ -15,7 +15,7 @@ def system_config_dir():
 
 
 def global_config_dir():
-    return os.getenv(env.DVC_SYSTEM_CONFIG_DIR) or platformdirs.user_config_dir(
+    return os.getenv(env.DVC_GLOBAL_CONFIG_DIR) or platformdirs.user_config_dir(
         APPNAME, APPAUTHOR
     )
 

dvc/dvcfile.py

@@ -1,5 +1,4 @@
 import contextlib
-import logging
 import os
 from typing import (
     TYPE_CHECKING,
@@ -14,6 +13,7 @@ from typing import (
 )
 
 from dvc.exceptions import DvcException
+from dvc.log import logger
 from dvc.stage import serialize
 from dvc.stage.exceptions import (
     StageFileBadNameError,
@@ -32,7 +32,7 @@ if TYPE_CHECKING:
     from .parsing import DataResolver
     from .stage import Stage
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 _T = TypeVar("_T")
 
 DVC_FILE_SUFFIX = ".dvc"
@@ -166,7 +166,6 @@ class FileMixin:
             **kwargs,
         )
 
-    # pylint: disable-next=unused-argument
     def remove(self, force=False):  # noqa: ARG002
         with contextlib.suppress(FileNotFoundError):
             os.unlink(self.path)
@@ -208,7 +207,6 @@ class SingleStageFile(FileMixin):
         dump_yaml(self.path, serialize.to_single_stage_file(stage, **kwargs))
         self.repo.scm_context.track_file(self.relpath)
 
-    # pylint: disable-next=unused-argument
     def remove_stage(self, stage):  # noqa: ARG002
         self.remove()
 
@@ -237,9 +235,7 @@ class ProjectFile(FileMixin):
         self.__dict__.pop("resolver", None)
         self.__dict__.pop("stages", None)
 
-    def dump(
-        self, stage, update_pipeline=True, update_lock=True, **kwargs
-    ):  # pylint: disable=arguments-differ
+    def dump(self, stage, update_pipeline=True, update_lock=True, **kwargs):
         """Dumps given stage appropriately in the dvcfile."""
         from dvc.stage import PipelineStage
 
@@ -298,7 +294,7 @@ class ProjectFile(FileMixin):
     def resolver(self) -> "DataResolver":
         from .parsing import DataResolver
 
-        wdir = self.repo.fs.path.parent(self.path)
+        wdir = self.repo.fs.parent(self.path)
         return DataResolver(self.repo, wdir, self.contents)
 
     @cached_property