testing: bench: use --dataset instead of --size (#10134)

This is no longer just about size (e.g. we have mnist/openorca/used-cars),
so the --dataset makes more sense.

.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/codeql.yml

@@ -34,6 +34,7 @@ jobs:
       uses: github/codeql-action/init@v2
       with:
         languages: ${{ matrix.language }}
+        queries: security-extended
 
     - name: Perform CodeQL Analysis
       uses: github/codeql-action/analyze@v2

.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.5'
     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
@@ -36,7 +36,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
@@ -65,9 +65,9 @@ def yappi_profile(
         else:
             if ctx_id is not None:
                 print(f"\nThread {ctx_id}")  # noqa: T201
-            st.print_all()  # pylint:disable=no-member
+            st.print_all()
             if ctx_id is None:
-                threads.print_all()  # pylint:disable=no-member
+                threads.print_all()
 
     yappi.clear_stats()
 
@@ -76,7 +76,7 @@ def yappi_profile(
 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
@@ -115,9 +115,9 @@ def profile(dump_path: Optional[str] = None):
 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 +145,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

@@ -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
 
@@ -27,6 +26,7 @@ from dvc.commands import (
     get_url,
     git_hook,
     imp,
+    imp_db,
     imp_url,
     init,
     install,
@@ -46,10 +46,11 @@ from dvc.commands import (
     update,
     version,
 )
+from dvc.log import logger
 
 from . import DvcParserError
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 COMMANDS = [
     init,
@@ -65,6 +66,7 @@ COMMANDS = [
     data_sync,
     gc,
     imp,
+    imp_db,
     imp_url,
     config,
     checkout,
@@ -95,12 +97,12 @@ COMMANDS = [
 
 
 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 +114,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.path.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/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
 

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_adapter
+        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_adapter(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(

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/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,6 +8,7 @@ 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
 
@@ -16,7 +16,7 @@ if TYPE_CHECKING:
     from dvc.fs import FileSystem
     from dvc.types import DictStrAny, StrPath
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class ConfigError(DvcException):
@@ -96,7 +96,7 @@ class Config(dict):
         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
@@ -390,10 +390,10 @@ class Config(dict):
 
 
 def _parse_named(conf):
-    result: Dict[str, Dict] = {"remote": {}, "machine": {}}
+    result: Dict[str, Dict] = {"remote": {}, "machine": {}, "db": {}}
 
     for section, val in conf.items():
-        match = re_find(r'^\s*(remote|machine)\s*"(.*)"\s*$', section)
+        match = re_find(r'^\s*(remote|machine|db)\s*"(.*)"\s*$', section)
         if match:
             key, name = match
             result[key][name] = val
@@ -408,7 +408,7 @@ def _pack_named(conf):
     result = compact(conf)
 
     # Transform remote.name -> 'remote "name"'
-    for key in ("remote", "machine"):
+    for key in ("remote", "machine", "db"):
         for name, val in conf[key].items():
             result[f'{key} "{name}"'] = val
         result.pop(key, None)

dvc/config_schema.py

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

dvc/daemon.py

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

dvc/dagascii.py

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

dvc/data_cloud.py

@@ -1,10 +1,10 @@
 """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 +15,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:

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
+
+Adapter = Union[sqla.SQLAlchemyAdapter, dbt_query.DbtAdapter]
+
+
+def get_adapter(
+    config,
+    project_dir: Optional[str] = None,
+    profile: Optional[str] = None,
+    target: Optional[str] = None,
+    **kwargs: Any,
+) -> "ContextManager[Adapter]":
+    if config:
+        return sqla.adapter(config, **kwargs)
+    return dbt_query.adapter(project_dir=project_dir, profile=profile, target=target)
+
+
+__all__ = ["export", "get_adapter", "get_model", "Adapter"]

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 DbtAdapter:
+    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 adapter(
+    project_dir: Optional[str] = None,
+    profiles_dir: Optional[str] = None,
+    profile: Optional[str] = None,
+    target: Optional[str] = None,
+) -> Iterator["DbtAdapter"]:
+    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 DbtAdapter(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 SQLAlchemyAdapter:
+    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 adapter(
+    url_or_config: Union[str, URL, Dict[str, str]], **engine_kwargs: Any
+) -> Iterator[SQLAlchemyAdapter]:
+    url = url_from_config(url_or_config)
+    with handle_error(url):
+        engine = create_engine(url, **engine_kwargs)
+    try:
+        yield SQLAlchemyAdapter(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/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_adapter
+
+        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_adapter(
+            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):
@@ -132,8 +132,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
 

dvc/env.py

@@ -1,4 +1,6 @@
+DVC_ANALYTICS_ENDPOINT = "DVC_ANALYTICS_ENDPOINT"
 DVC_DAEMON = "DVC_DAEMON"
+DVC_DAEMON_LOGFILE = "DVC_DAEMON_LOGFILE"
 DVC_EXP_AUTO_PUSH = "DVC_EXP_AUTO_PUSH"
 DVC_EXP_BASELINE_REV = "DVC_EXP_BASELINE_REV"
 DVC_EXP_GIT_REMOTE = "DVC_EXP_GIT_REMOTE"
@@ -12,6 +14,7 @@ DVC_SHOW_TRACEBACK = "DVC_SHOW_TRACEBACK"
 DVC_SITE_CACHE_DIR = "DVC_SITE_CACHE_DIR"
 DVC_STUDIO_OFFLINE = "DVC_STUDIO_OFFLINE"
 DVC_STUDIO_REPO_URL = "DVC_STUDIO_REPO_URL"
-DVC_STUDIO_TOKEN = "DVC_STUDIO_TOKEN"  # noqa: S105 # nosec B105
+DVC_STUDIO_TOKEN = "DVC_STUDIO_TOKEN"  # noqa: S105
 DVC_STUDIO_URL = "DVC_STUDIO_URL"
 DVC_SYSTEM_CONFIG_DIR = "DVC_SYSTEM_CONFIG_DIR"
+DVC_UPDATER_ENDPOINT = "DVC_UPDATER_ENDPOINT"

dvc/exceptions.py

@@ -301,7 +301,7 @@ class URLMissingError(DvcException):
         super().__init__(f"The path '{url}' does not exist")
 
 
-class IsADirectoryError(DvcException):  # noqa: A001,pylint:disable=redefined-builtin
+class IsADirectoryError(DvcException):  # noqa: A001
     """Raised when a file operation is requested on a directory."""
 
 

dvc/fs/__init__.py

@@ -29,7 +29,7 @@ from dvc_objects.fs.path import Path  # noqa: F401
 
 from .callbacks import Callback
 from .data import DataFileSystem  # noqa: F401
-from .dvc import DVCFileSystem  # noqa: F401
+from .dvc import DVCFileSystem
 from .git import GitFileSystem  # noqa: F401
 
 known_implementations.update(
@@ -46,12 +46,11 @@ known_implementations.update(
 )
 
 
-# pylint: enable=unused-import
-
-
 def download(
     fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None
 ) -> int:
+    from dvc.scm import lfs_prefetch
+
     with Callback.as_tqdm_callback(
         desc=f"Downloading {fs.path.name(fs_path)}",
         unit="files",
@@ -75,6 +74,8 @@ def download(
             from_infos = [fs_path]
             to_infos = [to]
 
+        if isinstance(fs, DVCFileSystem):
+            lfs_prefetch(fs, from_infos)
         cb.set_size(len(from_infos))
         jobs = jobs or fs.jobs
         generic.copy(
@@ -166,8 +167,8 @@ def get_cloud_fs(repo_config, **kwargs):
         # should be treated as being a root path.
         fs_path = cls.root_marker
     else:
-        fs_path = cls._strip_protocol(url)  # pylint:disable=protected-access
+        fs_path = cls._strip_protocol(url)
 
-    extras = cls._get_kwargs_from_urls(url)  # pylint:disable=protected-access
+    extras = cls._get_kwargs_from_urls(url)
     conf = {**extras, **remote_conf}  # remote config takes priority
     return cls, conf, fs_path

dvc/fs/callbacks.py

@@ -1,4 +1,3 @@
-# pylint: disable=unused-import
 from contextlib import ExitStack
 from typing import TYPE_CHECKING, Any, BinaryIO, Dict, Optional, Union
 
@@ -72,7 +71,7 @@ class RichCallback(Callback):
         value: int = 0,
         progress: Optional["RichTransferProgress"] = None,
         desc: Optional[str] = None,
-        bytes: bool = False,  # noqa: A002, pylint: disable=redefined-builtin
+        bytes: bool = False,  # noqa: A002
         unit: Optional[str] = None,
         disable: bool = False,
         transient: bool = True,

dvc/fs/data.py

@@ -1,8 +1,8 @@
 import functools
-import logging
 import os
 from typing import TYPE_CHECKING
 
+from dvc.log import logger
 from dvc.utils import as_posix
 from dvc_objects.fs.base import FileSystem
 
@@ -10,7 +10,7 @@ if TYPE_CHECKING:
     from dvc_data.fs import DataFileSystem as _DataFileSystem
 
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 
 class DataFileSystem(FileSystem):
@@ -22,7 +22,7 @@ class DataFileSystem(FileSystem):
         return config
 
     @functools.cached_property
-    def fs(  # pylint: disable=invalid-overridden-method
+    def fs(
         self,
     ) -> "_DataFileSystem":
         from dvc_data.fs import DataFileSystem as _DataFileSystem

dvc/fs/dvc.py

@@ -1,6 +1,5 @@
 import errno
 import functools
-import logging
 import ntpath
 import os
 import posixpath
@@ -11,6 +10,7 @@ from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Tuple, Type, Un
 from fsspec.spec import AbstractFileSystem
 from funcy import wrap_with
 
+from dvc.log import logger
 from dvc_objects.fs.base import FileSystem
 from dvc_objects.fs.path import Path
 
@@ -20,7 +20,7 @@ if TYPE_CHECKING:
     from dvc.repo import Repo
     from dvc.types import DictStrAny, StrPath
 
-logger = logging.getLogger(__name__)
+logger = logger.getChild(__name__)
 
 RepoFactory = Union[Callable[..., "Repo"], Type["Repo"]]
 Key = Tuple[str, ...]
@@ -72,7 +72,7 @@ def _get_dvc_path(dvc_fs, subkey):
     return dvc_fs.path.join(*subkey) if subkey else ""
 
 
-class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
+class _DVCFileSystem(AbstractFileSystem):
     cachable = False
     root_marker = "/"
 
@@ -84,9 +84,7 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         subrepos: bool = False,
         repo_factory: Optional[RepoFactory] = None,
         fo: Optional[str] = None,
-        # pylint:disable-next=unused-argument
         target_options: Optional[Dict[str, Any]] = None,  # noqa: ARG002
-        # pylint:disable-next=unused-argument
         target_protocol: Optional[str] = None,  # noqa: ARG002
         config: Optional["DictStrAny"] = None,
         remote: Optional[str] = None,
@@ -146,7 +144,7 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
                 remote_config=remote_config,
             )
             assert repo is not None
-            # pylint: disable=protected-access
+
             repo_factory = repo._fs_conf["repo_factory"]
             self._repo_stack.enter_context(repo)
 
@@ -282,9 +280,7 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         dvc_fs = self._datafss.get(repo_key)
         return repo, dvc_fs, subkey
 
-    def _open(
-        self, path, mode="rb", **kwargs
-    ):  # pylint: disable=arguments-renamed, arguments-differ
+    def _open(self, path, mode="rb", **kwargs):
         if mode != "rb":
             raise OSError(errno.EROFS, os.strerror(errno.EROFS))
 
@@ -307,9 +303,7 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         except FileNotFoundError:
             return False
 
-    def ls(  # pylint: disable=arguments-differ # noqa: C901
-        self, path, detail=True, dvc_only=False, **kwargs
-    ):
+    def ls(self, path, detail=True, dvc_only=False, **kwargs):
         key = self._get_key_from_relative(path)
         repo, dvc_fs, subkey = self._get_subrepo_info(key)
 
@@ -320,7 +314,7 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
             with suppress(FileNotFoundError, NotADirectoryError):
                 for info in dvc_fs.ls(dvc_path, detail=True):
                     dvc_infos[dvc_fs.path.name(info["name"])] = info
-            dvc_exists = bool(dvc_infos) or dvc_fs.exists(dvc_path)
+                dvc_exists = True
 
         fs_exists = False
         fs_infos = {}
@@ -333,10 +327,13 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
                     fs, fs_path, detail=True, ignore_subrepos=ignore_subrepos
                 ):
                     fs_infos[fs.path.name(info["name"])] = info
+                fs_exists = True
             except (FileNotFoundError, NotADirectoryError):
                 pass
 
-            fs_exists = bool(fs_infos) or fs.exists(fs_path)
+        if not (dvc_exists or fs_exists):
+            # broken symlink or TreeError
+            raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), path)
 
         dvcfiles = kwargs.get("dvcfiles", False)
 
@@ -344,10 +341,6 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         paths = []
         names = set(dvc_infos.keys()) | set(fs_infos.keys())
 
-        if not names and (dvc_exists or fs_exists):
-            # broken symlink or TreeError
-            raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), path)
-
         for name in names:
             if not dvcfiles and _is_dvc_file(name):
                 continue
@@ -415,7 +408,7 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         info["name"] = path
         return info
 
-    def get_file(self, rpath, lpath, **kwargs):  # pylint: disable=arguments-differ
+    def get_file(self, rpath, lpath, **kwargs):
         key = self._get_key_from_relative(rpath)
         fs_path = self._from_key(key)
         try:
@@ -440,7 +433,6 @@ class DVCFileSystem(FileSystem):
         return config
 
     @functools.cached_property
-    # pylint: disable-next=invalid-overridden-method
     def fs(self) -> "_DVCFileSystem":
         return _DVCFileSystem(**self.fs_args)
 
@@ -452,7 +444,7 @@ class DVCFileSystem(FileSystem):
         return self.fs.isdvc(path, **kwargs)
 
     @property
-    def path(self) -> Path:  # pylint: disable=invalid-overridden-method
+    def path(self) -> Path:
         return self.fs.path
 
     @property

dvc/fs/git.py

@@ -10,7 +10,7 @@ if TYPE_CHECKING:
     from dvc.scm import Git
 
 
-class GitFileSystem(FileSystem):  # pylint:disable=abstract-method
+class GitFileSystem(FileSystem):
     """Proxies the repo file access methods to Git objects"""
 
     protocol = "local"
@@ -39,7 +39,7 @@ class GitFileSystem(FileSystem):  # pylint:disable=abstract-method
         )
 
     @functools.cached_property
-    def fs(  # pylint: disable=invalid-overridden-method
+    def fs(
         self,
     ) -> "FsspecGitFileSystem":
         from scmrepo.fs import GitFileSystem as FsspecGitFileSystem
@@ -47,7 +47,7 @@ class GitFileSystem(FileSystem):  # pylint:disable=abstract-method
         return FsspecGitFileSystem(**self.fs_args)
 
     @functools.cached_property
-    def path(self):  # pylint: disable=invalid-overridden-method
+    def path(self):
         return self.fs.path
 
     @property