feat: widen dbt-core compatibility range

Author: eakmanrq

.github/workflows/pr.yaml

@@ -8,6 +8,8 @@ on:
 concurrency:
   group: 'pr-${{ github.event.pull_request.number }}'
   cancel-in-progress: true
+permissions:
+  contents: read
 jobs:
   test-vscode:
     env:
@@ -66,3 +68,34 @@ jobs:
           name: playwright-report
           path: vscode/extension/playwright-report/
           retention-days: 30
+  test-dbt-versions:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        dbt-version: ["1.3.0", "1.4.0", "1.5.0", "1.6.0", "1.7.0", "1.8.0", "1.9.0", "1.10.0"]
+    steps:
+      - uses: actions/checkout@v5
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.10'
+      - name: Install SQLMesh dev dependencies
+        run: |
+          python -m venv .venv
+          source .venv/bin/activate
+          make install-dev
+      - name: Install dbt-core version
+        run: |
+          source .venv/bin/activate
+          pip install "dbt-core~=${{ matrix.dbt-version }}"
+          pip install pydantic>=2.0.0 --force-reinstall
+      - name: Run dbt tests
+        run: |
+          source .venv/bin/activate
+          make dbt-test
+      - name: Test SQLMesh info in sushi_dbt
+        working-directory: ./examples/sushi_dbt
+        run: |
+          source ../../.venv/bin/activate
+          sqlmesh info --skip-connection

sqlmesh/dbt/manifest.py

@@ -15,7 +15,12 @@
 from sqlmesh.utils.conversions import make_serializable
 
 # Override the file name to prevent dbt commands from invalidating the cache.
-dbt_constants.PARTIAL_PARSE_FILE_NAME = "sqlmesh_partial_parse.msgpack"
+if hasattr(dbt_constants, "PARTIAL_PARSE_FILE_NAME"):
+    dbt_constants.PARTIAL_PARSE_FILE_NAME = "sqlmesh_partial_parse.msgpack"
+else:
+    from dbt.parser import manifest as dbt_manifest
+
+    dbt_manifest.PARTIAL_PARSE_FILE_NAME = "sqlmesh_partial_parse.msgpack"
 
 import jinja2
 from dbt.adapters.factory import register_adapter, reset_adapters
@@ -377,11 +382,17 @@ def _load_on_run_start_end(self) -> None:
 
                 if "on-run-start" in node.tags:
                     self._on_run_start_per_package[node.package_name][node_name] = HookConfig(
-                        sql=sql, index=node.index or 0, path=node_path, dependencies=dependencies
+                        sql=sql,
+                        index=getattr(node, "index", None) or 0,
+                        path=node_path,
+                        dependencies=dependencies,
                     )
                 else:
                     self._on_run_end_per_package[node.package_name][node_name] = HookConfig(
-                        sql=sql, index=node.index or 0, path=node_path, dependencies=dependencies
+                        sql=sql,
+                        index=getattr(node, "index", None) or 0,
+                        path=node_path,
+                        dependencies=dependencies,
                     )
 
     @property
@@ -591,6 +602,9 @@ def _macro_references(
     manifest: Manifest, node: t.Union[ManifestNode, Macro]
 ) -> t.Set[MacroReference]:
     result: t.Set[MacroReference] = set()
+    if not hasattr(node, "depends_on"):
+        return result
+
     for macro_node_id in node.depends_on.macros:
         if not macro_node_id:
             continue
@@ -606,18 +620,20 @@ def _macro_references(
 
 def _refs(node: ManifestNode) -> t.Set[str]:
     if DBT_VERSION >= (1, 5, 0):
-        result = set()
+        result: t.Set[str] = set()
+        if not hasattr(node, "refs"):
+            return result
         for r in node.refs:
-            ref_name = f"{r.package}.{r.name}" if r.package else r.name
+            ref_name = f"{r.package}.{r.name}" if r.package else r.name  # type: ignore
             if getattr(r, "version", None):
-                ref_name = f"{ref_name}_v{r.version}"
+                ref_name = f"{ref_name}_v{r.version}"  # type: ignore
             result.add(ref_name)
         return result
     return {".".join(r) for r in node.refs}  # type: ignore
 
 
 def _sources(node: ManifestNode) -> t.Set[str]:
-    return {".".join(s) for s in node.sources}
+    return {".".join(s) for s in getattr(node, "sources", [])}
 
 
 def _model_node_id(model_name: str, package: str) -> str:

sqlmesh/dbt/seed.py

@@ -90,31 +90,33 @@ def to_sqlmesh(
         )
 
 
-class Integer(agate_helper.Integer):
-    def cast(self, d: t.Any) -> t.Optional[int]:
-        if isinstance(d, str):
-            # The dbt's implementation doesn't support coercion of strings to integers.
-            if d.strip().lower() in self.null_values:
-                return None
-            try:
-                return int(d)
-            except ValueError:
-                raise agate.exceptions.CastError('Can not parse value "%s" as Integer.' % d)
-        return super().cast(d)
-
-    def jsonify(self, d: t.Any) -> str:
-        return d
-
-
-agate_helper.Integer = Integer  # type: ignore
-
-
 AGATE_TYPE_MAPPING = {
-    agate_helper.Integer: exp.DataType.build("int"),
     agate_helper.Number: exp.DataType.build("double"),
     agate_helper.ISODateTime: exp.DataType.build("datetime"),
     agate.Date: exp.DataType.build("date"),
     agate.DateTime: exp.DataType.build("datetime"),
     agate.Boolean: exp.DataType.build("boolean"),
     agate.Text: exp.DataType.build("text"),
 }
+
+
+if hasattr(agate_helper, "Integer"):
+
+    class Integer(agate_helper.Integer):
+        def cast(self, d: t.Any) -> t.Optional[int]:
+            if isinstance(d, str):
+                # The dbt's implementation doesn't support coercion of strings to integers.
+                if d.strip().lower() in self.null_values:
+                    return None
+                try:
+                    return int(d)
+                except ValueError:
+                    raise agate.exceptions.CastError('Can not parse value "%s" as Integer.' % d)
+            return super().cast(d)
+
+        def jsonify(self, d: t.Any) -> str:
+            return d
+
+    agate_helper.Integer = Integer  # type: ignore
+
+    AGATE_TYPE_MAPPING[agate_helper.Integer] = exp.DataType.build("int")

tests/dbt/test_integration.py

@@ -7,7 +7,7 @@
 
 import pandas as pd  # noqa: TID253
 import pytest
-from dbt.cli.main import dbtRunner
+from dbt.cli.main import dbtRunner  # type: ignore
 import time_machine
 
 from sqlmesh import Context