diff --git a/dvc/commands/config.py b/dvc/commands/config.py
index ff8461bc6..6cd50a780 100644
--- a/dvc/commands/config.py
+++ b/dvc/commands/config.py
@@ -1,6 +1,8 @@
 import argparse
 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
@@ -8,7 +10,7 @@ from dvc.ui import ui
 
 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")
 
diff --git a/dvc/commands/imp_db.py b/dvc/commands/imp_db.py
index 97faf430e..fe1f47ed6 100644
--- a/dvc/commands/imp_db.py
+++ b/dvc/commands/imp_db.py
@@ -1,13 +1,68 @@
 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 not 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):
@@ -24,6 +79,7 @@ class CmdImportDb(CmdBase):
             output_format=self.args.output_format,
             out=self.args.out,
             force=self.args.force,
+            connection=self.args.connection,
         )
         return 0
 
@@ -86,5 +142,25 @@ def add_parser(subparsers, parent_parser):
         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_parser = subparsers.add_parser(
+        "test-db",
+        parents=[parent_parser],
+        description=append_doc_link(IMPORT_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)
diff --git a/dvc/config.py b/dvc/config.py
index f95b1f20f..998ddca96 100644
--- a/dvc/config.py
+++ b/dvc/config.py
@@ -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)
diff --git a/dvc/config_schema.py b/dvc/config_schema.py
index 16d3e9af0..5870d2d69 100644
--- a/dvc/config_schema.py
+++ b/dvc/config_schema.py
@@ -318,8 +318,8 @@ SCHEMA = {
     "feature": FeatureSchema(
         {
             Optional("machine", default=False): Bool,
-            "db_profile": str,
-            "db_target": str,
+            "dbt_profile": str,
+            "dbt_target": str,
         },
     ),
     "plots": {
@@ -352,4 +352,11 @@ SCHEMA = {
         "repo_url": str,
         Optional("offline", default=False): Bool,
     },
+    "db": {
+        str: {
+            "url": str,
+            "username": str,
+            "password": str,
+        },
+    },
 }
diff --git a/dvc/database/__init__.py b/dvc/database/__init__.py
new file mode 100644
index 000000000..4d1ad9f6a
--- /dev/null
+++ b/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"]
diff --git a/dvc/database/dbt_models.py b/dvc/database/dbt_models.py
new file mode 100644
index 000000000..347c44009
--- /dev/null
+++ b/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)
diff --git a/dvc/database/dbt_query.py b/dvc/database/dbt_query.py
new file mode 100644
index 000000000..19c8cf1ae
--- /dev/null
+++ b/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)
diff --git a/dvc/database/dbt_runner.py b/dvc/database/dbt_runner.py
new file mode 100644
index 000000000..3413d9dc7
--- /dev/null
+++ b/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
diff --git a/dvc/database/dbt_utils.py b/dvc/database/dbt_utils.py
new file mode 100644
index 000000000..87111ddb1
--- /dev/null
+++ b/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)
diff --git a/dvc/database/serializer.py b/dvc/database/serializer.py
new file mode 100644
index 000000000..62f5ba40d
--- /dev/null
+++ b/dvc/database/serializer.py
@@ -0,0 +1,48 @@
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Union
+
+if TYPE_CHECKING:
+    import sqlalchemy as sa
+    from agate import Table
+
+
+@dataclass
+class PandasSQLSerializer:
+    sql: "Union[sa.TextClause, str]"
+    con: "sa.Connection"
+    chunksize: int = 10_000
+
+    def to_csv(self, file: str) -> 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)
+
+    def to_json(self, file: str) -> None:
+        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) -> None:
+        return self.table.to_csv(file)
+
+    def to_json(self, file: str) -> None:
+        return self.table.to_json(file)
+
+
+def export(
+    serializer: Union[PandasSQLSerializer, AgateSerializer],
+    file: str,
+    format: str = "csv",  # noqa: A002
+) -> None:
+    if format == "csv":
+        return serializer.to_csv(file)
+    return serializer.to_json(file)
diff --git a/dvc/database/sqla.py b/dvc/database/sqla.py
new file mode 100644
index 000000000..9d61515e3
--- /dev/null
+++ b/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()
diff --git a/dvc/dependency/db.py b/dvc/dependency/db.py
index a72ee4cda..d99ea933b 100644
--- a/dvc/dependency/db.py
+++ b/dvc/dependency/db.py
@@ -1,6 +1,6 @@
 import os
 from contextlib import contextmanager, nullcontext
-from typing import TYPE_CHECKING, Any, Dict, Iterator, Union
+from typing import TYPE_CHECKING, Any, Dict, Iterator, Optional, Union
 
 from funcy import compact
 
@@ -11,9 +11,9 @@ from dvc.scm import SCM
 from .base import Dependency
 
 if TYPE_CHECKING:
-    from agate import Table
     from rich.status import Status
 
+    from dvc.output import Output
     from dvc.stage import Stage
 
 logger = logger.getChild(__name__)
@@ -24,20 +24,23 @@ PARAM_PROFILE = "profile"
 PARAM_FILE_FORMAT = "file_format"
 
 
-def _get_db_config(config: Dict) -> Dict:
+def _get_dbt_config(config: Dict) -> Dict:
     conf = config.get("feature", {})
-    pref = "db_"
+    pref = "dbt_"
     return {k.lstrip(pref): v for k, v in conf.items() if k.startswith(pref)}
 
 
 @contextmanager
-def log_status(msg, log=logger.debug) -> Iterator["Status"]:
+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), ui.status(msg) as status:
-        yield status
+    with log_durations(log, msg), status or ui.status(msg) as st:
+        st.update(msg)
+        yield st
 
 
 @contextmanager
@@ -50,15 +53,10 @@ def chdir(path):
         os.chdir(wdir)
 
 
-def export_to(table: "Table", to: str, file_format: str = "csv") -> None:
-    exporter = {"csv": table.to_csv, "json": table.to_json}
-    return exporter[file_format](to)
-
-
 class AbstractDependency(Dependency):
     """Dependency without workspace/fs/fs_path"""
 
-    def __init__(self, stage: "Stage", info, *args, **kwargs):
+    def __init__(self, stage: "Stage", info: Dict[str, Any], *args, **kwargs):
         self.repo = stage.repo
         self.stage = stage
         self.fs = None
@@ -69,7 +67,8 @@ class AbstractDependency(Dependency):
 
 class DbDependency(AbstractDependency):
     PARAM_QUERY = "query"
-    QUERY_SCHEMA = {PARAM_QUERY: str}
+    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)
@@ -103,33 +102,37 @@ class DbDependency(AbstractDependency):
     def update(self, rev=None):
         """nothing to update."""
 
-    def download(self, to, jobs=None, file_format=None):  # noqa: ARG002
+    def download(self, to, jobs=None, file_format=None, **kwargs):  # noqa: ARG002
+        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")
 
-        from dvc.utils.db import _check_dbt, _profiles_dir, execute_sql
+        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")
 
-        db_config = _get_db_config(self.repo.config)
-        profile = db_info.get(PARAM_PROFILE) or db_config.get(PARAM_PROFILE)
-        target = self.target or db_config.get("target")
-        file_format = file_format or db_info.get(PARAM_FILE_FORMAT, "csv")
+        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")
 
-        _check_dbt(self.PARAM_QUERY)
-        profiles_dir = _profiles_dir(self.repo.root_dir)
-        with log_status("Executing query") as status:
-            table = execute_sql(
-                query,
-                profiles_dir,
-                self.repo.root_dir,
-                profile,
-                target=target,
-                status=status,
-            )
-        # NOTE: we keep everything in memory, and then export it out later.
-        with log_status(f"Saving to {to}"):
-            return export_to(table, to.fs_path, file_format)
+        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")
+            with log_status("Executing query") as status, db.query(query) as serializer:
+                logger.debug("using serializer: %s", serializer)
+                with log_status(f"Saving to {to}", status=status):
+                    return export(serializer, to.fs_path, format=file_format)
 
 
 class DbtDependency(AbstractDependency):
@@ -237,7 +240,12 @@ class DbtDependency(AbstractDependency):
             rev = self.rev
         self.def_repo[RepoDependency.PARAM_REV_LOCK] = self._get_clone(rev).get_rev()
 
-    def download(self, to, jobs=None, file_format=None):  # noqa: ARG002
+    def download(
+        self,
+        to: "Output",
+        jobs: Optional[int] = None,  # noqa: ARG002
+        file_format: Optional[str] = None,
+    ) -> None:
         from dvc.ui import ui
 
         from .repo import RepoDependency
@@ -257,25 +265,33 @@ class DbtDependency(AbstractDependency):
             self._download_db(to, file_format=file_format)
         ui.write(f"Saved file to {to}", styled=True)
 
-    def _download_db(self, to, version=None, file_format=None):
-        from dvc.utils.db import get_model
+    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")
 
-        db_config = _get_db_config(self.repo.config)
-        version = version or db_info.get(self.PARAM_VERSION)
-        profile = db_info.get(PARAM_PROFILE) or db_config.get(PARAM_PROFILE)
-        target = self.target or db_info.get("target")
-        file_format = file_format or db_info.get(PARAM_FILE_FORMAT, "csv")
+        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"):
-            table = get_model(model, version=version, profile=profile, target=target)
+            serializer = get_model(
+                model, version=version, profile=profile, target=target
+            )
         # NOTE: we keep everything in memory, and then export it out later.
         with log_status(f"Saving to {to}"):
-            export_to(table, to.fs_path, file_format=file_format)
+            export(serializer, to.fs_path, format=file_format)
 
 
 DB_SCHEMA = {
diff --git a/dvc/repo/imp_db.py b/dvc/repo/imp_db.py
index eb7cd235f..00f494b99 100644
--- a/dvc/repo/imp_db.py
+++ b/dvc/repo/imp_db.py
@@ -27,6 +27,7 @@ def imp_db(  # noqa: PLR0913
     output_format: str = "csv",
     out: Optional[str] = None,
     force: bool = False,
+    connection: Optional[str] = None,
 ):
     ui.warn("WARNING: import-db is an experimental feature.")
     ui.warn(
@@ -50,7 +51,7 @@ def imp_db(  # noqa: PLR0913
         db.update({"model": model, "version": version, "project_dir": project_dir})
     else:
         out = out or f"results.{output_format}"
-        db["query"] = sql
+        db.update({"query": sql, "connection": connection})
 
     out = resolve_output(url or ".", out, force=force)
     path, wdir, out = resolve_paths(self, out, always_local=True)
diff --git a/dvc/utils/db.py b/dvc/utils/db.py
deleted file mode 100644
index 00e75c381..000000000
--- a/dvc/utils/db.py
+++ /dev/null
@@ -1,252 +0,0 @@
-import os
-from contextlib import contextmanager
-from importlib.util import find_spec
-from typing import (
-    TYPE_CHECKING,
-    List,
-    Optional,
-    Sequence,
-    Union,
-)
-
-from funcy import cut_prefix, identity
-
-from dvc.exceptions import DvcException
-from dvc.log import logger
-
-from . import packaging
-
-if TYPE_CHECKING:
-    from agate import Table
-    from dbt.config.profile import Profile
-    from dbt.contracts.results import RunResult
-    from rich.status import Status
-
-logger = logger.getChild(__name__)
-
-
-class DbtInternalError(DvcException):
-    pass
-
-
-def _check_dbt(action: Optional[str]):
-    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")
-    packaging.check_required_version(pkg="dbt-core")
-
-
-@contextmanager
-def check_dbt(action: Optional[str]):
-    _check_dbt(action)
-    yield
-
-
-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 + ") }}"
-
-
-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
-
-
-@check_dbt("model")
-def get_model(
-    name: str,
-    package: Optional[str] = None,
-    version: Optional[int] = None,
-    profile: Optional[str] = None,
-    target: Optional[str] = None,
-) -> "Table":
-    model = _ref(name, package, version=version)
-    q = f"select * from {model}"  # noqa: S608
-    return _dbt_show(
-        inline=q,
-        profile=profile,
-        target=target,
-    )
-
-
-def _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 _global_dbt_flags(
-    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_profile_or(
-    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 os.path.isfile(os.path.join(project_dir, "dbt_project.yml")):
-        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)
-
-
-@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
-
-
-@contextmanager
-def _suppress_status(status: Optional["Status"]):
-    if status:
-        status.stop()
-
-    yield
-    if status:
-        status.start()
-
-
-@check_dbt("query")
-def execute_sql(
-    sql: str,
-    profiles_dir: str,
-    project_dir: Optional[str],
-    profile: Optional[str],
-    target: Optional[str] = None,
-    status: Optional["Status"] = None,
-) -> "Table":
-    from dbt.adapters import factory as adapters_factory
-    from dbt.adapters.sql import SQLAdapter
-
-    flags = _global_dbt_flags(profiles_dir, os.getcwd(), target=target)
-    update_status = status.update if status else identity
-
-    with flags, adapters_factory.adapter_management():
-        # likely invalid connection profile or no adapter
-        with _handle_profile_parsing_error(), _suppress_status(status):
-            profile_obj = _get_profile_or(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, SQLAdapter)
-        with adapter.connection_named("debug"):
-            update_status("Testing Connection")
-            try:
-                adapter.debug_query()
-            except Exception as exc:
-                if status:
-                    status.stop()
-                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
-
-        with adapter.connection_named("execute"):
-            update_status("Executing query")
-            exec_resp = adapter.execute(sql, fetch=True)
-            _, table = exec_resp
-            return table
diff --git a/pyproject.toml b/pyproject.toml
index eec2bd86d..0403b2d7e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -77,6 +77,7 @@ gs = ["dvc-gs==2.22.1"]
 hdfs = ["dvc-hdfs==2.19"]
 lint = [
     "mypy==1.7.0",
+    "pandas-stubs",
     "types-colorama",
     "types-psutil",
     "types-pyinstaller",
@@ -111,6 +112,8 @@ tests = [
     'pywin32>=225; sys_platform == "win32"', # optional test dependency
     'tzdata; sys_platform == "win32"',  # for testing with celery
     "dbt-sqlite", # optional dependency for testing `import-db`
+    "sqlalchemy>=1,<3", # optional dependency for `import-db`
+    "pandas>=1",
 ]
 webdav = ["dvc-webdav==2.19.1"]
 webhdfs = ["dvc-webhdfs==2.19"]
diff --git a/tests/func/test_cli.py b/tests/func/test_cli.py
index 9f4cc1ba9..fbcf45e36 100644
--- a/tests/func/test_cli.py
+++ b/tests/func/test_cli.py
@@ -96,7 +96,7 @@ def test_config_unset(dvc):
     cmd = args.func(args)
     assert isinstance(cmd, CmdConfig)
     assert args.unset
-    assert args.name == (False, "section", "option")
+    assert args.name == (None, "section", "option")
     assert args.value == value
 
 
diff --git a/tests/func/test_import_db.py b/tests/func/test_import_db.py
index bfe35401c..e0d390be8 100644
--- a/tests/func/test_import_db.py
+++ b/tests/func/test_import_db.py
@@ -1,13 +1,25 @@
 """import_db tests without involving any databases."""
-
+import pytest
 from agate import Table
 
+from dvc.database.serializer import AgateSerializer
+
 data1 = Table([(i, i) for i in range(1, 10)], ["id", "value"])
 data2 = Table([(i, i) for i in range(10, 20)], ["id", "value"])
 
+serializers = AgateSerializer(data1), AgateSerializer(data2)
 
-def test_sql(mocker, tmp_dir, dvc):
-    mocker.patch("dvc.utils.db.execute_sql", side_effect=[data1, data2])
+
+@pytest.fixture
+def adapter(mocker):
+    m = mocker.patch("dvc.database.get_adapter")
+    adapter = mocker.MagicMock()
+    adapter.query.return_value.__enter__.side_effect = serializers
+    m.return_value.__enter__.return_value = adapter
+    return m
+
+
+def test_sql(adapter, tmp_dir, dvc):
     stage = dvc.imp_db(
         sql="select * from model", profile="profile", output_format="json"
     )
@@ -44,8 +56,46 @@ def test_sql(mocker, tmp_dir, dvc):
     }
 
 
+def test_sql_conn_string(adapter, tmp_dir, dvc):
+    with dvc.config.edit(level="local") as conf:
+        conf["db"] = {"conn": {"url": "conn"}}
+
+    stage = dvc.imp_db(
+        sql="select * from model", connection="conn", output_format="json"
+    )
+    db = {"file_format": "json", "connection": "conn", "query": "select * from model"}
+    assert (tmp_dir / stage.relpath).parse() == {
+        "md5": "f0dcca8e8411907adb25aa547d7c432c",
+        "frozen": True,
+        "deps": [{"db": db}],
+        "outs": [
+            {
+                "hash": "md5",
+                "md5": "56226d443067b195c90b427db557e1f2",
+                "path": "results.json",
+                "size": 243,
+            }
+        ],
+    }
+
+    dvc.update(stage.addressing)
+    assert (tmp_dir / stage.relpath).parse() == {
+        "md5": "8c0441b206d8ee7c7b9e3d54871c3cbb",
+        "frozen": True,
+        "deps": [{"db": db}],
+        "outs": [
+            {
+                "hash": "md5",
+                "md5": "08d806ee7e1f309c4561750805f02276",
+                "path": "results.json",
+                "size": 290,
+            }
+        ],
+    }
+
+
 def test_model(mocker, tmp_dir, dvc):
-    mocker.patch("dvc.utils.db.get_model", side_effect=[data1, data2])
+    mocker.patch("dvc.database.get_model", side_effect=serializers)
     stage = dvc.imp_db(model="model", output_format="json")
 
     assert (tmp_dir / stage.relpath).parse() == {
diff --git a/tests/func/test_import_db_e2e.py b/tests/func/test_import_db_e2e.py
index ae9108000..f0934d5d6 100644
--- a/tests/func/test_import_db_e2e.py
+++ b/tests/func/test_import_db_e2e.py
@@ -12,10 +12,6 @@ from agate import Table
 
 from dvc.types import StrPath
 
-dbt_connections = pytest.importorskip(
-    "dbt.adapters.sql.connections", reason="dbt-core not installed"
-)
-SQLConnectionManager = dbt_connections.SQLConnectionManager
 pytest.importorskip("dbt.adapters.sqlite", reason="dbt-sqlite not installed")
 
 
@@ -52,6 +48,14 @@ def seed_db(db_path):
         yield inner
 
 
+@pytest.fixture
+def db_config(dvc, db_path):
+    # only needed for `sql_conn` tests
+    with dvc.config.edit(level="local") as conf:
+        conf["db"] = {"conn": {"url": f"sqlite:///{db_path.fs_path}"}}
+    return "conn"
+
+
 @pytest.fixture
 def dbt_profile(tmp_dir, scm, db_path):
     (tmp_dir / "profiles.yml").dump(
@@ -90,11 +94,14 @@ def dbt_model(dbt_project):
     return "model"
 
 
-@pytest.fixture(params=("sql", "model", "external_model"))
+@pytest.fixture(params=("sql", "model", "external_model", "sql_conn"))
 def import_db_parameters(request: pytest.FixtureRequest):
     if request.param == "sql":
         profile = request.getfixturevalue("dbt_profile")
         return {"sql": "select * from model", "profile": profile}
+    if request.param == "sql_conn":
+        conn = request.getfixturevalue("db_config")
+        return {"sql": "select * from model", "connection": conn}
 
     dbt_project = request.getfixturevalue("dbt_project")
     return {