Feat: support dynamic blueprinting (#3997)

Author: georgesittas

sqlmesh/core/model/definition.py

@@ -1928,7 +1928,7 @@ def load_sql_based_models(
     **loader_kwargs: t.Any,
 ) -> t.List[Model]:
     gateway: t.Optional[exp.Expression] = None
-    blueprints: t.Optional[t.List[t.Optional[exp.Expression]]] = None
+    blueprints: t.Optional[exp.Expression] = None
 
     model_meta = seq_get(expressions, 0)
     for prop in (isinstance(model_meta, d.Model) and model_meta.expressions) or []:
@@ -1937,6 +1937,22 @@ def load_sql_based_models(
         elif prop.name == "blueprints":
             blueprints = prop.args["value"]
 
+    if isinstance(blueprints, d.MacroFunc):
+        rendered_blueprints = render_expression(
+            expression=blueprints,
+            module_path=module_path,
+            macros=loader_kwargs.get("macros"),
+            jinja_macros=loader_kwargs.get("jinja_macros"),
+            variables=get_variables(None),
+            path=path,
+            dialect=dialect,
+            default_catalog=loader_kwargs.get("default_catalog"),
+        )
+        if not rendered_blueprints:
+            raise_config_error("Failed to render blueprints property", path)
+
+        blueprints = t.cast(t.List, rendered_blueprints)[0]
+
     return create_models_from_blueprints(
         gateway=gateway,
         blueprints=blueprints,

tests/core/test_model.py

@@ -8233,6 +8233,49 @@ def entrypoint(evaluator):
     )
 
 
+def test_dynamic_blueprinting(tmp_path: Path) -> None:
+    init_example_project(tmp_path, dialect="duckdb", template=ProjectTemplate.EMPTY)
+
+    dynamic_template = tmp_path / "models/dynamic_template.sql"
+    dynamic_template.parent.mkdir(parents=True, exist_ok=True)
+    dynamic_template.write_text(
+        """
+        MODEL (
+          name @customer.some_table,
+          kind FULL,
+          blueprints @gen_blueprints(),
+        );
+
+        SELECT
+          @field_a,
+          @{field_b} AS field_b
+        FROM @customer.some_source
+
+        """
+    )
+
+    gen_blueprints = tmp_path / "macros/gen_blueprints.py"
+    gen_blueprints.parent.mkdir(parents=True, exist_ok=True)
+    gen_blueprints.write_text(
+        """from sqlmesh import macro
+
+@macro()
+def gen_blueprints(evaluator):
+    return (
+        "((customer := customer1, field_a := x, field_b := y),"
+        " (customer := customer2, field_a := z, field_b := w))"
+    )"""
+    )
+
+    ctx = Context(
+        config=Config(model_defaults=ModelDefaultsConfig(dialect="duckdb")), paths=tmp_path
+    )
+
+    assert len(ctx.models) == 2
+    assert '"memory"."customer1"."some_table"' in ctx.models
+    assert '"memory"."customer2"."some_table"' in ctx.models
+
+
 @time_machine.travel("2020-01-01 00:00:00 UTC")
 def test_dynamic_date_spine_model(assert_exp_eq):
     @macro()