fix: treat Python model gateways as literals

Signed-off-by: Joseph Finlayson <joseph.finlayson@gmail.com>

Author: josephfinlayson

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"),

tests/core/test_model.py

@@ -5039,6 +5039,38 @@ 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",