Fix: [dbt] materialized can be jinja and allow, but warn, on extra profile fields (#721)

* [dbt] materialized field can be jinja

* [dbt] Warn on extra profile fields

Author: crericha

sqlmesh/dbt/basemodel.py

@@ -259,15 +259,27 @@ def _context_for_dependencies(
     ) -> DbtContext:
         model_context = context.copy()
 
-        model_context.sources = {
-            name: value for name, value in context.sources.items() if name in dependencies.sources
-        }
-        model_context.seeds = {
-            name: value for name, value in context.seeds.items() if name in dependencies.refs
-        }
-        model_context.models = {
-            name: value for name, value in context.models.items() if name in dependencies.refs
-        }
+        models = {}
+        seeds = {}
+        sources = {}
+
+        for ref in self._dependencies.refs:
+            if ref in context.seeds:
+                seeds[ref] = context.seeds[ref]
+            elif ref in context.models:
+                models[ref] = context.models[ref]
+            else:
+                raise ConfigError(f"Model '{ref}' was not found for model '{self.table_name}'.")
+
+        for source in self._dependencies.sources:
+            if source in context.sources:
+                sources[source] = context.sources[source]
+            else:
+                raise ConfigError(f"Source '{source}' was not found for model '{self.table_name}'.")
+
+        model_context.sources = sources
+        model_context.seeds = seeds
+        model_context.models = models
         model_context.variables = {
             name: value
             for name, value in context.variables.items()
@@ -365,16 +377,8 @@ def _extract_value(node: t.Any) -> t.Any:
         return config
 
     def _ref(self, package_name: str, model_name: t.Optional[str] = None) -> BaseRelation:
-        if package_name in self.context.models:
-            relation = BaseRelation.create(**self.context.models[package_name].relation_info)
-        elif package_name in self.context.seeds:
-            relation = BaseRelation.create(**self.context.seeds[package_name].relation_info)
-        else:
-            raise ConfigError(
-                f"Model '{package_name}' was not found for model '{self.config.table_name}'."
-            )
         self._captured_dependencies.refs.add(package_name)
-        return relation
+        return BaseRelation.create()
 
     def _var(self, name: str, default: t.Optional[str] = None) -> t.Any:
         if default is None and name not in self.context.variables:
@@ -386,12 +390,8 @@ def _var(self, name: str, default: t.Optional[str] = None) -> t.Any:
 
     def _source(self, source_name: str, table_name: str) -> BaseRelation:
         full_name = ".".join([source_name, table_name])
-        if full_name not in self.context.sources:
-            raise ConfigError(
-                f"Source '{full_name}' was not found for model '{self.config.table_name}'."
-            )
         self._captured_dependencies.sources.add(full_name)
-        return BaseRelation.create(**self.context.sources[full_name].relation_info)
+        return BaseRelation.create()
 
     class TrackingAdapter(ParsetimeAdapter):
         def __init__(self, outer_self: ModelSqlRenderer, *args: t.Any, **kwargs: t.Any):

sqlmesh/dbt/loader.py

@@ -56,9 +56,6 @@ def _load_models(
         context = project.context.copy()
 
         for package_name, package in project.packages.items():
-            context.add_sources(package.sources)
-            context.add_seeds(package.seeds)
-            context.add_models(package.models)
             context.jinja_macros.add_macros(
                 package.macros,
                 package=package_name if package_name != context.project_name else None,

sqlmesh/dbt/model.py

@@ -70,7 +70,7 @@ class ModelConfig(BaseModelConfig):
     start: t.Optional[str] = None
     cluster_by: t.Optional[t.List[str]] = None
     incremental_strategy: t.Optional[str] = None
-    materialized: Materialization = Materialization.VIEW
+    materialized: str = Materialization.VIEW.value
     sql_header: t.Optional[str] = None
     unique_key: t.Optional[t.List[str]] = None
 
@@ -94,10 +94,6 @@ def _validate_list(cls, v: t.Union[str, t.List[str]]) -> t.List[str]:
     def _validate_sql(cls, v: t.Union[str, SqlStr]) -> SqlStr:
         return SqlStr(v)
 
-    @validator("materialized", pre=True)
-    def _validate_materialization(cls, v: str) -> Materialization:
-        return Materialization(v.lower())
-
     _FIELD_UPDATE_STRATEGY: t.ClassVar[t.Dict[str, UpdateStrategy]] = {
         **BaseModelConfig._FIELD_UPDATE_STRATEGY,
         **{
@@ -112,15 +108,15 @@ def model_dialect(self) -> t.Optional[str]:
 
     @property
     def model_materialization(self) -> Materialization:
-        return self.materialized
+        return Materialization(self.materialized.lower())
 
     def model_kind(self, target: TargetConfig) -> ModelKind:
         """
         Get the sqlmesh ModelKind
         Returns:
             The sqlmesh ModelKind
         """
-        materialization = self.materialized
+        materialization = self.model_materialization
         if materialization == Materialization.TABLE:
             return ModelKind(name=ModelKindName.FULL)
         if materialization == Materialization.VIEW:

sqlmesh/dbt/project.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 import typing as t
 from pathlib import Path
 
@@ -9,6 +10,8 @@
 from sqlmesh.dbt.profile import Profile
 from sqlmesh.utils.errors import ConfigError
 
+logger = logging.getLogger(__name__)
+
 
 class Project:
     """Configuration for a DBT project"""
@@ -64,6 +67,13 @@ def load(cls, context: DbtContext) -> Project:
         profile = Profile.load(context, context.target_name)
         context.target = profile.target
 
+        extra_fields = profile.target.extra
+        if extra_fields:
+            extra_str = ",".join(f"'{field}'" for field in extra_fields)
+            logger.warning(
+                f"Warning: {profile.target.type} adapter does not currently support {extra_str}."
+            )
+
         packages = {}
         root_loader = PackageLoader(context, ProjectConfig())
 

sqlmesh/dbt/target.py

@@ -16,10 +16,9 @@
     SnowflakeConnectionConfig,
 )
 from sqlmesh.core.model import IncrementalByTimeRangeKind, IncrementalByUniqueKeyKind
-from sqlmesh.dbt.common import QuotingConfig
+from sqlmesh.dbt.common import DbtConfig, QuotingConfig
 from sqlmesh.utils import AttributeDict
 from sqlmesh.utils.errors import ConfigError
-from sqlmesh.utils.pydantic import PydanticModel
 
 if sys.version_info >= (3, 9):
     from typing import Literal
@@ -29,7 +28,7 @@
 IncrementalKind = t.Union[t.Type[IncrementalByUniqueKeyKind], t.Type[IncrementalByTimeRangeKind]]
 
 
-class TargetConfig(abc.ABC, PydanticModel):
+class TargetConfig(abc.ABC, DbtConfig):
     """
     Configuration for DBT profile target
 
@@ -92,6 +91,10 @@ def target_jinja(self, profile_name: str) -> AttributeDict:
     def quoting(self) -> QuotingConfig:
         return QuotingConfig()
 
+    @property
+    def extra(self) -> t.Set[str]:
+        return self.extra_fields(set(self.dict()))
+
 
 class DuckDbConfig(TargetConfig):
     """

tests/dbt/test_transformation.py

@@ -218,16 +218,17 @@ def test_config_containing_missing_dependency():
     context = DbtContext()
     model = ModelConfig(sql="{{ config(pre_hook=\"{{ print(ref('bar')) }}\") }} SELECT 1 FROM a")
     with pytest.raises(ConfigError, match="'bar' was not found"):
-        model.render_config(context)
+        model.render_config(context).to_sqlmesh(context)
 
     model = ModelConfig(sql='{{ config(pre_hook="{{ get_table_name() }}") }} SELECT 1 FROM a')
     with pytest.raises(ConfigError, match="get_table_name"):
-        model.render_config(context)
+        model.render_config(context).to_sqlmesh(context)
 
     model = ModelConfig(sql="{{ config(alias='{{ get_table_name() }}') }} SELECT 1 FROM a")
     rendered = model.render_config(context)
     assert rendered.alias == "{{ get_table_name() }}"
     assert "get_table_name" not in [macro.name for macro in rendered._dependencies.macros]
+    rendered.to_sqlmesh(context)
 
 
 def test_config_containing_method():
@@ -273,6 +274,16 @@ def test_schema_jinja(sushi_test_project: Project):
     model_config.to_sqlmesh(context).render_query().sql() == "SELECT 1 AS one FROM sushi AS sushi"
 
 
+def test_materialized_jinja(sushi_test_project: Project):
+    model_config = ModelConfig(
+        materialized="{{ 'table' if target.type in ['duckdb'] else 'view' }}",
+        sql="SELECT 1 AS one FROM {{ schema }}",
+    )
+    context = sushi_test_project.context
+    rendered = model_config.render_config(context)
+    assert rendered.materialized == "table"
+
+
 def test_this(assert_exp_eq, sushi_test_project: Project):
     model_config = ModelConfig(alias="test", sql="SELECT 1 AS one FROM {{ this.identifier }}")
     context = sushi_test_project.context