Fix: Support blueprint variables in jinja macros (#4289)

Author: izeigerman

docs/concepts/macros/jinja_macros.md

@@ -122,6 +122,31 @@ Access gateway variables in models using the same methods as [global variables](
 
 Gateway-specific variable values take precedence over variables with the same name specified in the configuration file's root `variables` key.
 
+### Blueprint variables
+
+Blueprint variables are defined as a property of the `MODEL` statement, and serve as a mechanism for [creating model templates](../models/sql_models.md):
+
+```sql linenums="1"
+MODEL (
+  name @customer.some_table,
+  kind FULL,
+  blueprints (
+    (customer := customer1, field_a := x, field_b := y),
+    (customer := customer2, field_a := z)
+  )
+);
+
+JINJA_QUERY_BEGIN;
+SELECT
+  {{ blueprint_var('field_a') }}
+  {{ blueprint_var('field_b', 'default_b') }} AS field_b
+FROM {{ blueprint_var('customer') }}.some_source
+JINJA_END;
+```
+
+Blueprint variables can be accessed using the `{{ blueprint_var() }}` macro function, which also supports specifying default values in case the variable is undefined (similar to `{{ var() }}`).
+
+
 ### Local variables
 
 Define your own variables with the Jinja statement `{% set ... %}`. For example, we could specify the name of the `num_orders` column in the `sqlmesh_example.full_model` like this:

sqlmesh/core/constants.py

@@ -79,6 +79,7 @@
 SQLMESH_BLUEPRINT_VARS = "__sqlmesh__blueprint__vars__"
 
 VAR = "var"
+BLUEPRINT_VAR = "blueprint_var"
 GATEWAY = "gateway"
 
 SQLMESH_MACRO = "__sqlmesh__macro__"

sqlmesh/utils/jinja.py

@@ -15,6 +15,7 @@
 from sqlmesh.core import dialect as d
 from sqlmesh.utils import AttributeDict
 from sqlmesh.utils.pydantic import PRIVATE_FIELDS, PydanticModel, field_serializer, field_validator
+from sqlmesh.utils.metaprogramming import SqlValue
 
 
 if t.TYPE_CHECKING:
@@ -593,7 +594,10 @@ def jinja_call_arg_name(node: nodes.Node) -> str:
 
 def create_var(variables: t.Dict[str, t.Any]) -> t.Callable:
     def _var(var_name: str, default: t.Optional[t.Any] = None) -> t.Optional[t.Any]:
-        return variables.get(var_name.lower(), default)
+        value = variables.get(var_name.lower(), default)
+        if isinstance(value, SqlValue):
+            return value.sql
+        return value
 
     return _var
 
@@ -603,10 +607,12 @@ def create_builtin_globals(
 ) -> t.Dict[str, t.Any]:
     global_vars.pop(c.GATEWAY, None)
     variables = global_vars.pop(c.SQLMESH_VARS, None) or {}
+    blueprint_variables = global_vars.pop(c.SQLMESH_BLUEPRINT_VARS, None) or {}
     return {
+        **global_vars,
         c.VAR: create_var(variables),
         c.GATEWAY: lambda: variables.get(c.GATEWAY, None),
-        **global_vars,
+        c.BLUEPRINT_VAR: create_var(blueprint_variables),
     }
 
 

tests/core/test_model.py

@@ -8662,6 +8662,55 @@ def test_blueprint_variable_precedence_sql(tmp_path: Path, assert_exp_eq: t.Call
     )
 
 
+def test_blueprint_variable_jinja(tmp_path: Path, assert_exp_eq: t.Callable) -> None:
+    init_example_project(tmp_path, dialect="duckdb", template=ProjectTemplate.EMPTY)
+
+    blueprint_variables = tmp_path / "models/blueprint_variables.sql"
+    blueprint_variables.parent.mkdir(parents=True, exist_ok=True)
+    blueprint_variables.write_text(
+        """
+        MODEL (
+          name s.@{bp_name},
+          blueprints (
+            (bp_name := m1, var1 := 'v1', var2 := v2),
+            (bp_name := m2, var1 := 'v3'),
+          ),
+        );
+
+        @DEF(bp_name, override);
+
+        JINJA_QUERY_BEGIN;
+        SELECT
+            {{ blueprint_var('var1') }} AS var1,
+            '{{ blueprint_var('var2') }}' AS var2,
+            '{{ blueprint_var('var2', 'var2_default') }}' AS var2_default,
+            '{{ blueprint_var('bp_name') }}' AS bp_name
+        FROM s.{{ blueprint_var('bp_name') }}_source;
+        JINJA_END;
+        """
+    )
+    ctx = Context(
+        config=Config(
+            model_defaults=ModelDefaultsConfig(dialect="duckdb"),
+            variables={"var2": "1"},
+        ),
+        paths=tmp_path,
+    )
+    assert len(ctx.models) == 2
+
+    m1 = ctx.get_model("s.m1", raise_if_missing=True)
+    m2 = ctx.get_model("s.m2", raise_if_missing=True)
+
+    assert_exp_eq(
+        m1.render_query(),
+        """SELECT 'v1' AS "var1", 'v2' AS "var2", 'v2' AS "var2_default", 'm1' AS "bp_name" FROM "memory"."s"."m1_source" AS "m1_source" """,
+    )
+    assert_exp_eq(
+        m2.render_query(),
+        """SELECT 'v3' AS "var1", 'None' AS "var2", 'var2_default' AS "var2_default", 'm2' AS "bp_name" FROM "memory"."s"."m2_source" AS "m2_source" """,
+    )
+
+
 def test_blueprint_variable_precedence_python(tmp_path: Path, mocker: MockerFixture) -> None:
     init_example_project(tmp_path, dialect="duckdb", template=ProjectTemplate.EMPTY)