fix(adapter): name dbt v2 as the cause when the dbt import fails

recce/adapter/dbt_adapter/__init__.py

@@ -27,6 +27,7 @@
 
 from recce.event import log_performance
 from recce.exceptions import (
+    DbtUnavailableError,
     DuckDBExternalAccessBlocked,
     RecceException,
     UnsupportedDbtSchemaError,
@@ -50,9 +51,16 @@
     import dbt.adapters.factory
     from dbt.contracts.state import PreviousState
 except ImportError as e:
-    print("Error: dbt module not found. Please install it by running:")
-    print("pip install dbt-core dbt-<adapter>")
-    raise e
+    import importlib.util
+
+    if importlib.util.find_spec("dbt") is None:
+        raise DbtUnavailableError(
+            "dbt module not found. Please install it by running:\n  pip install dbt-core dbt-<adapter>"
+        ) from e
+    # dbt v2 (Fusion) has a dbt._core module and no dbt.adapters module.
+    if importlib.util.find_spec("dbt._core") is not None:
+        raise DbtUnavailableError("Recce supports dbt-core 1.x. dbt v2 (Fusion) is not supported yet.") from e
+    raise
 
 try:
     from dbt.artifacts.exceptions import IncompatibleSchemaError
@@ -237,17 +245,14 @@ def as_manifest(m: WritableManifest) -> Manifest:
         return result
 
 
-# Highest schema versions dbt 1.x ever emits; dbt v2 / Fusion jumps straight to
-# v20, so anything above these is a Fusion artifact. Do not tie the ceiling to
-# what the installed dbt is compatible with: under dbt 1.6 a v12 manifest is
-# already incompatible, but it comes from dbt 1.x and should surface dbt's own
-# version-mismatch error, not be reported as Fusion.
+# dbt-core 1.x and dbt v2 both write manifest v12 and catalog v1.
+# Keep these limits fixed. Do not set them from the installed dbt: an old dbt,
+# such as dbt 1.7, cannot read a v12 manifest, and that case must raise dbt's
+# own IncompatibleSchemaError, not UnsupportedDbtSchemaError.
 _DBT1X_MAX_SCHEMA = {"manifest": 12, "catalog": 1}
 
 
 def _guard_unsupported_schema(artifact: str, found_version_url):
-    """Raise UnsupportedDbtSchemaError if `found_version_url` is a dbt v2 / Fusion
-    schema (above the dbt 1.x ceiling); no-op otherwise."""
     # dbt_schema_version looks like "https://schemas.getdbt.com/dbt/manifest/v12.json"
     match = re.search(r"/v(\d+)\.json", str(found_version_url))
     found = int(match.group(1)) if match else None

recce/exceptions.py

@@ -21,16 +21,19 @@ def __str__(self):
         return super().__str__()
 
 
+class DbtUnavailableError(RecceException):
+    """Raised when the dbt adapter cannot import dbt: dbt is not installed, or dbt v2 (Fusion) is installed."""
+
+
 class UnsupportedDbtSchemaError(RecceException):
-    """Raised when a dbt artifact's schema is newer than the bundled dbt 1.x
-    supports, i.e. a dbt v2 / Fusion artifact. See _DBT1X_MAX_SCHEMA in the dbt
-    adapter for the version ceiling."""
+    """Raised when a dbt artifact's schema is newer than Recce supports.
+    The version limits are in _DBT1X_MAX_SCHEMA in the dbt adapter."""
 
     def __init__(self, artifact: str, found_version: int):
         message = (
-            f"dbt v2 / Fusion {artifact}s (schema v{found_version}) are not yet "
-            f"supported by Recce. Recce supports dbt 1.x artifacts (manifest schema up "
-            f"to v12, catalog v1). Re-generate the {artifact} with dbt 1.x."
+            f"The {artifact} uses schema v{found_version}, which is newer than Recce "
+            f"supports (manifest up to v12, catalog v1). Re-generate the {artifact} "
+            f"with dbt-core 1.x."
         )
         super().__init__(message, is_raise=True)
 

recce/server.py

@@ -46,7 +46,7 @@
 )
 from .core import RecceContext, default_context, load_context
 from .event import get_recce_api_token, log_api_event, log_single_env_event
-from .exceptions import RecceException
+from .exceptions import DbtUnavailableError, RecceException
 from .github import is_github_codespace
 from .models.types import CllData
 from .models.websocket import CloudUserContextMessage
@@ -256,6 +256,9 @@ def _do_lifespan_setup(app_state: AppState):
 
 @asynccontextmanager
 async def lifespan(fastapi: FastAPI):
+    from rich.console import Console
+    from rich.markup import escape
+
     from recce.core import default_context
     from recce.event import log_performance
     from recce.util.startup_perf import clear_startup_tracker, get_startup_tracker
@@ -295,7 +298,13 @@ async def background_load():
                         tracker.node_count = len(recce_ctx.adapter.curr_manifest.nodes)
                 log_performance("server_startup", tracker.to_dict())
         except Exception as e:
-            logger.exception("Failed to load server context during startup")
+            if isinstance(e, DbtUnavailableError):
+                logger.debug("Failed to load server context during startup", exc_info=True)
+                console = Console(stderr=True)
+                message = escape(str(e))
+                console.print(f"[[red]Error[/red]] Failed to load server context during startup: {message}")
+            else:
+                logger.exception("Failed to load server context during startup")
             app_state.startup_error = e
         finally:
             clear_startup_tracker()

tests/test_dbt.py

@@ -1,5 +1,7 @@
 import json
 import os
+import subprocess
+import sys
 import tempfile
 from unittest import TestCase
 from unittest.mock import MagicMock
@@ -12,9 +14,12 @@
 current_dir = os.path.dirname(os.path.abspath(__file__))
 
 
-def _fusion_artifact(kind: str) -> dict:
+_ABOVE_CEILING = {"manifest": 13, "catalog": 2}
+
+
+def _newer_artifact(kind: str) -> dict:
     return {
-        "metadata": {"dbt_schema_version": f"https://schemas.getdbt.com/dbt/{kind}/v20.json"},
+        "metadata": {"dbt_schema_version": f"https://schemas.getdbt.com/dbt/{kind}/v{_ABOVE_CEILING[kind]}.json"},
     }
 
 
@@ -47,58 +52,51 @@ def test_load_lineage_with_catalog(self):
         assert len(lineage["nodes"]["model.jaffle_shop.orders"]["columns"]) == 9
 
 
-class TestFusionManifestFailLoud(TestCase):
-    """A v20 (dbt v2 / Fusion) artifact must fail loud with a Recce-branded message."""
-
-    def _assert_friendly(self, exc_info):
+class TestNewerSchemaFailLoud(TestCase):
+    def _assert_friendly(self, exc_info, kind):
         msg = str(exc_info.value)
-        assert "Fusion" in msg
-        assert "v20" in msg
-        assert "not yet supported" in msg
+        assert f"v{_ABOVE_CEILING[kind]}" in msg
+        assert "newer than Recce supports" in msg
 
-    def test_load_manifest_v20_data_fails_loud(self):
+    def test_load_manifest_newer_data_fails_loud(self):
         with pytest.raises(UnsupportedDbtSchemaError) as exc_info:
-            load_manifest(data=_fusion_artifact("manifest"))
-        self._assert_friendly(exc_info)
+            load_manifest(data=_newer_artifact("manifest"))
+        self._assert_friendly(exc_info, "manifest")
 
-    def test_load_manifest_v20_path_fails_loud(self):
+    def test_load_manifest_newer_path_fails_loud(self):
         with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
-            json.dump(_fusion_artifact("manifest"), f)
+            json.dump(_newer_artifact("manifest"), f)
             path = f.name
         try:
             with pytest.raises(UnsupportedDbtSchemaError) as exc_info:
                 load_manifest(path=path)
-            self._assert_friendly(exc_info)
+            self._assert_friendly(exc_info, "manifest")
         finally:
             os.unlink(path)
 
-    def test_load_catalog_v20_data_fails_loud(self):
+    def test_load_catalog_newer_data_fails_loud(self):
         with pytest.raises(UnsupportedDbtSchemaError) as exc_info:
-            load_catalog(data=_fusion_artifact("catalog"))
-        self._assert_friendly(exc_info)
+            load_catalog(data=_newer_artifact("catalog"))
+        self._assert_friendly(exc_info, "catalog")
 
-    def test_load_catalog_v20_path_fails_loud(self):
+    def test_load_catalog_newer_path_fails_loud(self):
         with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
-            json.dump(_fusion_artifact("catalog"), f)
+            json.dump(_newer_artifact("catalog"), f)
             path = f.name
         try:
             with pytest.raises(UnsupportedDbtSchemaError) as exc_info:
                 load_catalog(path=path)
-            self._assert_friendly(exc_info)
+            self._assert_friendly(exc_info, "catalog")
         finally:
             os.unlink(path)
 
-    def test_v12_not_flagged_as_fusion(self):
-        # A v12 manifest / v1 catalog is dbt 1.x, not Fusion — the guard must not fire,
-        # regardless of which dbt version Recce is running against.
+    def test_supported_schema_not_flagged(self):
         from recce.adapter.dbt_adapter import _guard_unsupported_schema
 
         _guard_unsupported_schema("manifest", "https://schemas.getdbt.com/dbt/manifest/v12.json")
         _guard_unsupported_schema("catalog", "https://schemas.getdbt.com/dbt/catalog/v1.json")
 
     def test_old_incompatible_artifact_keeps_dbt_error(self):
-        # v1 / v0 sit below the 1.x ceiling but dbt still rejects them as too old —
-        # the guard must stay silent, not mislabel them as Fusion.
         from recce.adapter.dbt_adapter import IncompatibleSchemaError
 
         for loader, kind, version in [(load_manifest, "manifest", 1), (load_catalog, "catalog", 0)]:
@@ -113,3 +111,34 @@ def test_old_incompatible_artifact_keeps_dbt_error(self):
                     loader(path=path)
             finally:
                 os.unlink(path)
+
+
+def _import_dbt_adapter(fake_packages_dir) -> subprocess.CompletedProcess:
+    pythonpath = os.pathsep.join(p for p in [str(fake_packages_dir), os.environ.get("PYTHONPATH")] if p)
+    return subprocess.run(
+        [sys.executable, "-c", "import recce.adapter.dbt_adapter"],
+        env={**os.environ, "PYTHONPATH": pythonpath},
+        capture_output=True,
+        text=True,
+    )
+
+
+def test_dbt_v2_import_names_dbt_v2(tmp_path):
+    (tmp_path / "dbt").mkdir()
+    (tmp_path / "dbt" / "__init__.py").write_text("")
+    (tmp_path / "dbt" / "_core.py").write_text("")
+
+    result = _import_dbt_adapter(tmp_path)
+
+    assert result.returncode != 0
+    assert "DbtUnavailableError: Recce supports dbt-core 1.x. dbt v2 (Fusion) is not supported yet." in result.stderr
+
+
+def test_unexpected_import_error_keeps_original_error(tmp_path):
+    (tmp_path / "agate.py").write_text("raise ImportError(\"cannot import name 'MappedSequence' from 'agate'\")")
+
+    result = _import_dbt_adapter(tmp_path)
+
+    assert result.returncode != 0
+    assert "ImportError: cannot import name 'MappedSequence' from 'agate'" in result.stderr
+    assert "DbtUnavailableError" not in result.stderr