Feat: Support gateway in model defaults

docs/guides/multi_engine.md

@@ -11,9 +11,13 @@ SQLMesh enables this decoupling by supporting multiple engine adapters within a
 Configuring your project to use multiple engines follows a simple process:
 
 - Include all required [gateway connections](../reference/configuration.md#connection) in your configuration.
-- Specify the `gateway` to be used for execution in the `MODEL` DDL.
+- Set `model_defaults.gateway` to the gateway most models should use, and override individual models
+  with `gateway` in the `MODEL` DDL when needed.
 
-If no gateway is explicitly defined for a model, the [default_gateway](../reference/configuration.md#default-gateway) of the project is used.
+If no gateway is explicitly defined for a model, SQLMesh uses the project's
+[`model_defaults.gateway`](../reference/model_configuration.md#model-defaults), when configured,
+and otherwise uses its [default_gateway](../reference/configuration.md#default-gateway). This lets
+all managed models in a project use a gateway without repeating it in every model definition.
 
 By default, virtual layer views are created in the `default_gateway`. This approach requires that all engines can read from and write to the same shared catalog, so a view in the `default_gateway` can access a table in another gateway.
 

docs/reference/model_configuration.md

@@ -193,10 +193,15 @@ The SQLMesh project-level `model_defaults` key supports the following options, d
 - allow_partials
 - enabled
 - interval_unit
+- gateway
 - pre_statements (described [here](../concepts/models/sql_models.md#pre--and-post-statements))
 - post_statements (described [here](../concepts/models/sql_models.md#pre--and-post-statements))
 - on_virtual_update (described [here](../concepts/models/sql_models.md#on-virtual-update-statements))
 
+The `gateway` default applies to managed SQL, Python, and seed models. It does not apply to
+external models because an external model's `gateway` selects a gateway-specific source
+definition. Set that gateway explicitly in `external_models.yaml` when needed.
+
 
 ### Model Naming
 

sqlmesh/core/config/model.py

@@ -50,6 +50,7 @@ class ModelDefaultsConfig(BaseConfig):
         pre_statements: The list of SQL statements that get executed before a model runs.
         post_statements: The list of SQL statements that get executed before a model runs.
         on_virtual_update: The list of SQL statements to be executed after the virtual update.
+        gateway: The gateway used by models that do not specify one explicitly.
 
     """
 
@@ -76,6 +77,7 @@ class ModelDefaultsConfig(BaseConfig):
     pre_statements: t.Optional[t.List[t.Union[str, exp.Expr]]] = None
     post_statements: t.Optional[t.List[t.Union[str, exp.Expr]]] = None
     on_virtual_update: t.Optional[t.List[t.Union[str, exp.Expr]]] = None
+    gateway: t.Optional[str] = None
 
     _model_kind_validator = model_kind_validator
     _on_destructive_change_validator = on_destructive_change_validator

sqlmesh/core/model/decorator.py

@@ -125,8 +125,12 @@ def models(
 
             blueprints = blueprints[0]
 
+        gateway = self.kwargs.get("gateway")
+        if isinstance(gateway, str) and gateway.lstrip().startswith("@"):
+            gateway = parse_one(gateway, dialect=dialect)
+
         return create_models_from_blueprints(
-            gateway=self.kwargs.get("gateway"),
+            gateway=gateway,
             blueprints=blueprints,
             get_variables=get_variables,
             loader=self.model,

sqlmesh/core/model/definition.py

@@ -2071,9 +2071,14 @@ def create_models_from_blueprints(
         loader_kwargs["default_catalog"] = original_default_catalog
         blueprint_variables = _extract_blueprint_variables(blueprint, path)
 
-        if gateway:
+        gateway_name: t.Optional[str]
+        if isinstance(gateway, str):
+            # Python decorator gateway names are literals, not SQL expressions. In particular,
+            # parsing a gateway such as "secondary-gw" as SQL would interpret it as subtraction.
+            gateway_name = gateway.lower()
+        elif gateway:
             rendered_gateway = render_expression(
-                expression=exp.maybe_parse(gateway, dialect=dialect),
+                expression=gateway,
                 module_path=module_path,
                 macros=loader_kwargs.get("macros"),
                 jinja_macros=loader_kwargs.get("jinja_macros"),
@@ -2082,7 +2087,11 @@ def create_models_from_blueprints(
                 default_catalog=loader_kwargs.get("default_catalog"),
                 blueprint_variables=blueprint_variables,
             )
-            gateway_name = rendered_gateway[0].name if rendered_gateway else None
+            gateway_name = rendered_gateway[0].name.lower() if rendered_gateway else None
+        elif configured_gateway := (loader_kwargs.get("defaults") or {}).get("gateway"):
+            # Config gateway names are literals, not SQL expressions. In particular, parsing a
+            # gateway such as "secondary-gw" as SQL would interpret it as subtraction.
+            gateway_name = configured_gateway.lower()
         else:
             gateway_name = None
 
@@ -2600,6 +2609,11 @@ def _create_model(
         kwargs["kind"] = create_model_kind(raw_kind, dialect, defaults or {})
 
     defaults = {k: v for k, v in (defaults or {}).items() if k in klass.all_fields()}
+    if issubclass(klass, ExternalModel):
+        # An external model's gateway selects a gateway-specific source definition in
+        # external_models.yaml, so it must remain explicit rather than inheriting the
+        # gateway used to execute managed models in the project.
+        defaults.pop("gateway", None)
     if not issubclass(klass, SqlModel):
         defaults.pop("optimize_query", None)
 

tests/core/integration/test_multi_repo.py

@@ -120,6 +120,71 @@ def test_multi(mocker):
     ]
 
 
+def test_multi_repo_model_default_gateways(tmp_path: Path) -> None:
+    """Each project routes its models using its own default gateway."""
+    repo_one = tmp_path / "repo_one"
+    repo_two = tmp_path / "repo_two"
+    (repo_one / "models").mkdir(parents=True)
+    (repo_two / "models").mkdir(parents=True)
+
+    (repo_one / "models" / "default.sql").write_text(
+        "MODEL (name analytics.repo_one_default, kind FULL); SELECT @owner AS owner"
+    )
+    (repo_one / "models" / "override.sql").write_text(
+        "MODEL (name analytics.explicit_override, kind FULL, gateway 'repo-two'); "
+        "SELECT @owner AS owner"
+    )
+    (repo_two / "models" / "default.sql").write_text(
+        "MODEL (name analytics.repo_two_default, kind FULL); SELECT @owner AS owner"
+    )
+
+    def make_config(project: str, default_gateway: str) -> Config:
+        return Config(
+            project=project,
+            gateways={
+                "repo-one": GatewayConfig(
+                    connection=DuckDBConnectionConfig(database=str(tmp_path / "repo_one.duckdb")),
+                    variables={"owner": "repo-one-variable"},
+                ),
+                "repo-two": GatewayConfig(
+                    connection=DuckDBConnectionConfig(database=str(tmp_path / "repo_two.duckdb")),
+                    variables={"owner": "repo-two-variable"},
+                ),
+            },
+            default_gateway=default_gateway,
+            model_defaults=ModelDefaultsConfig(dialect="duckdb", gateway=default_gateway),
+            variables={"owner": "global-variable"},
+        )
+
+    context = Context(
+        paths=[repo_one, repo_two],
+        config={
+            repo_one: make_config("repo_one", "repo-one"),
+            repo_two: make_config("repo_two", "repo-two"),
+        },
+        gateway="repo-one",
+    )
+
+    repo_one_model = context.get_model("repo_one.analytics.repo_one_default")
+    repo_two_model = context.get_model("repo_two.analytics.repo_two_default")
+    override_model = context.get_model("repo_two.analytics.explicit_override")
+
+    assert repo_one_model.gateway == "repo-one"
+    assert repo_one_model.catalog == "repo_one"
+    assert repo_one_model.project == "repo_one"
+    assert context.render(repo_one_model.fqn).sql() == ("SELECT 'repo-one-variable' AS \"owner\"")
+
+    assert repo_two_model.gateway == "repo-two"
+    assert repo_two_model.catalog == "repo_two"
+    assert repo_two_model.project == "repo_two"
+    assert context.render(repo_two_model.fqn).sql() == ("SELECT 'repo-two-variable' AS \"owner\"")
+
+    assert override_model.gateway == "repo-two"
+    assert override_model.catalog == "repo_two"
+    assert override_model.project == "repo_one"
+    assert context.render(override_model.fqn).sql() == ("SELECT 'repo-two-variable' AS \"owner\"")
+
+
 @use_terminal_console
 def test_multi_repo_single_project_environment_statements_update(copy_to_temp_path):
     paths = copy_to_temp_path("examples/multi")

tests/core/test_config.py

@@ -966,6 +966,27 @@ def test_gateway_model_defaults(tmp_path):
     assert ctx.config.model_defaults == expected
 
 
+def test_model_defaults_gateway_from_yaml(tmp_path: Path) -> None:
+    config_path = tmp_path / "config.yaml"
+    config_path.write_text(
+        """
+gateways:
+  project_gateway:
+    connection:
+      type: duckdb
+
+model_defaults:
+  dialect: duckdb
+  gateway: project_gateway
+""",
+        encoding="utf-8",
+    )
+
+    config = load_config_from_paths(Config, project_paths=[config_path])
+
+    assert config.model_defaults.gateway == "project_gateway"
+
+
 def test_model_defaults_cron_tz(tmp_path):
     """Test that cron_tz can be set in model_defaults."""
     import zoneinfo

tests/core/test_model.py

@@ -5039,6 +5039,62 @@ def python_model_prop(context, **kwargs):
     assert m.interval_unit == IntervalUnit.QUARTER_HOUR
 
 
+def test_explicit_hyphenated_gateway_python_model() -> None:
+    @model(
+        name="model_schema.python_explicit_gateway",
+        kind="full",
+        gateway="secondary-gw",
+        columns={"some_col": "int"},
+    )
+    def python_explicit_gateway(context, **kwargs):
+        yield {"some_col": 1}
+
+    requested_variable_gateways: t.List[t.Optional[str]] = []
+
+    def get_variables(gateway: t.Optional[str]) -> t.Dict[str, str]:
+        requested_variable_gateways.append(gateway)
+        return {}
+
+    loaded_models = model.get_registry()["model_schema.python_explicit_gateway"].models(
+        get_variables=get_variables,
+        module_path=Path("."),
+        path=Path("."),
+        dialect="duckdb",
+        defaults=ModelDefaultsConfig().dict(),
+        default_catalog="default_db",
+        default_catalog_per_gateway={"secondary-gw": "secondary_db"},
+    )
+
+    assert len(loaded_models) == 1
+    assert loaded_models[0].gateway == "secondary-gw"
+    assert loaded_models[0].catalog == "secondary_db"
+    assert requested_variable_gateways == ["secondary-gw"]
+
+
+def test_model_defaults_gateway_python_model() -> None:
+    @model(
+        name="model_schema.python_gateway_default",
+        kind="full",
+        columns={"some_col": "int"},
+    )
+    def python_gateway_default(context, **kwargs):
+        yield {"some_col": 1}
+
+    loaded_models = model.get_registry()["model_schema.python_gateway_default"].models(
+        get_variables=lambda gateway: {},
+        module_path=Path("."),
+        path=Path("."),
+        dialect="duckdb",
+        defaults=ModelDefaultsConfig(gateway="python_gateway").dict(),
+        default_catalog="default_db",
+        default_catalog_per_gateway={"python_gateway": "python_db"},
+    )
+
+    assert len(loaded_models) == 1
+    assert loaded_models[0].gateway == "python_gateway"
+    assert loaded_models[0].catalog == "python_db"
+
+
 def test_model_defaults_macros(make_snapshot):
     model_defaults = ModelDefaultsConfig(
         table_format="@IF(@gateway = 'dev', 'iceberg', NULL)",
@@ -12982,6 +13038,99 @@ def test_default_catalog_still_applied_to_supported_gateway():
     assert model.catalog == "other_db", f"Expected catalog 'other_db', got: {model.catalog}"
 
 
+@pytest.mark.parametrize(
+    ("model_gateway", "expected_gateway", "expected_catalog"),
+    [
+        (None, "secondary-gw", "secondary_db"),
+        ("default_gw", "default_gw", "example_catalog"),
+    ],
+)
+def test_model_defaults_gateway(
+    model_gateway: t.Optional[str], expected_gateway: str, expected_catalog: str
+) -> None:
+    """A project-level gateway default controls loading unless the model overrides it."""
+    gateway_property = f"gateway {model_gateway}," if model_gateway else ""
+    expressions = d.parse(
+        f"""
+        MODEL (
+            name my_schema.my_model,
+            kind FULL,
+            {gateway_property}
+        );
+
+        SELECT 1 AS id
+        """,
+        default_dialect="duckdb",
+    )
+    requested_variable_gateways: t.List[t.Optional[str]] = []
+
+    def get_variables(gateway: t.Optional[str]) -> t.Dict[str, str]:
+        requested_variable_gateways.append(gateway)
+        return {}
+
+    models = load_sql_based_models(
+        expressions,
+        get_variables=get_variables,
+        defaults=ModelDefaultsConfig(gateway="secondary-gw").dict(),
+        dialect="duckdb",
+        default_catalog_per_gateway={
+            "default_gw": "example_catalog",
+            "secondary-gw": "secondary_db",
+        },
+        default_catalog="example_catalog",
+    )
+
+    assert len(models) == 1
+    assert models[0].gateway == expected_gateway
+    assert models[0].catalog == expected_catalog
+    assert requested_variable_gateways == [expected_gateway]
+
+
+def test_model_defaults_gateway_with_blueprints() -> None:
+    expressions = d.parse(
+        """
+        MODEL (
+            name model_@suffix.my_model,
+            kind FULL,
+            blueprints (
+                (suffix := one),
+                (suffix := two),
+            ),
+        );
+
+        SELECT 1 AS id
+        """,
+        default_dialect="duckdb",
+    )
+
+    models = load_sql_based_models(
+        expressions,
+        get_variables=lambda gateway: {},
+        defaults=ModelDefaultsConfig(gateway="other_duckdb").dict(),
+        dialect="duckdb",
+        default_catalog_per_gateway={"other_duckdb": "other_db"},
+        default_catalog="example_catalog",
+    )
+
+    assert {model.gateway for model in models} == {"other_duckdb"}
+    assert {model.catalog for model in models} == {"other_db"}
+
+
+def test_external_model_does_not_inherit_model_defaults_gateway() -> None:
+    default_external_model = create_external_model(
+        "source_schema.default_source",
+        defaults=ModelDefaultsConfig(gateway="managed_gateway").dict(),
+    )
+    explicit_external_model = create_external_model(
+        "source_schema.explicit_source",
+        defaults=ModelDefaultsConfig(gateway="managed_gateway").dict(),
+        gateway="source_gateway",
+    )
+
+    assert default_external_model.gateway is None
+    assert explicit_external_model.gateway == "source_gateway"
+
+
 def test_no_gateway_uses_global_default_catalog():
     """
     Control test: when a model does NOT specify a gateway, the global