diff --git a/recce/adapter/dbt_adapter/__init__.py b/recce/adapter/dbt_adapter/__init__.py index 3314c8a85..1b4e7c765 100644 --- a/recce/adapter/dbt_adapter/__init__.py +++ b/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-") - 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-" + ) 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 diff --git a/recce/exceptions.py b/recce/exceptions.py index c953d6400..9939716e4 100644 --- a/recce/exceptions.py +++ b/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) diff --git a/recce/server.py b/recce/server.py index b97d98540..2f14e2eec 100644 --- a/recce/server.py +++ b/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() diff --git a/tests/test_dbt.py b/tests/test_dbt.py index d5cb01088..bab6b80d0 100644 --- a/tests/test_dbt.py +++ b/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