fix: rework dbt grants to sqlmesh grants converstion

- sqlmesh grants uses None it explicitly means unmanaged
  and {} to mean managed with no grants (revoke all existing grants)
- empty {} dbt grants config is always considered as unmanaged
  dbt manifest always parses None grants as {}

Author: newtonapple

sqlmesh/core/engine_adapter/base.py

@@ -2458,7 +2458,7 @@ def _apply_grants_config_expr(
         table: exp.Table,
         grant_config: GrantsConfig,
         table_type: DataObjectType = DataObjectType.TABLE,
-    ) -> t.List[exp.Grant]:
+    ) -> t.List[exp.Expression]:
         """Returns SQLGlot Grant expressions to apply grants to a table.
 
         Args:
@@ -2467,7 +2467,7 @@ def _apply_grants_config_expr(
             table_type: The type of database object (TABLE, VIEW, MATERIALIZED_VIEW).
 
         Returns:
-            List of SQLGlot Grant expressions.
+            List of SQLGlot expressions for grant operations.
 
         Raises:
             NotImplementedError: If the engine does not support grants.

sqlmesh/core/engine_adapter/postgres.py

@@ -145,7 +145,7 @@ def _dcl_grants_config_expr(
         dcl_cmd: t.Type[DCL],
         relation: exp.Expression,
         grant_config: GrantsConfig,
-    ) -> t.Union[t.List[exp.Grant], t.List[exp.Revoke]]:
+    ) -> t.List[exp.Expression]:
         expressions = []
         for privilege, principals in grant_config.items():
             if not principals:
@@ -154,23 +154,20 @@ def _dcl_grants_config_expr(
             grant = dcl_cmd(
                 privileges=[exp.GrantPrivilege(this=exp.Var(this=privilege))],
                 securable=relation,
-                principals=principals,  # use original strings so user can to choose quote or not
+                principals=principals,  # use original strings; no quoting
             )
             expressions.append(grant)
 
-        return expressions
+        return t.cast(t.List[exp.Expression], expressions)
 
     def _apply_grants_config_expr(
         self,
         table: exp.Table,
         grant_config: GrantsConfig,
         table_type: DataObjectType = DataObjectType.TABLE,
-    ) -> t.List[exp.Grant]:
+    ) -> t.List[exp.Expression]:
         # https://www.postgresql.org/docs/current/sql-grant.html
-        return t.cast(
-            t.List[exp.Grant],
-            self._dcl_grants_config_expr(exp.Grant, table, grant_config),
-        )
+        return self._dcl_grants_config_expr(exp.Grant, table, grant_config)
 
     def _revoke_grants_config_expr(
         self,
@@ -179,10 +176,7 @@ def _revoke_grants_config_expr(
         table_type: DataObjectType = DataObjectType.TABLE,
     ) -> t.List[exp.Expression]:
         # https://www.postgresql.org/docs/current/sql-revoke.html
-        return t.cast(
-            t.List[exp.Expression],
-            self._dcl_grants_config_expr(exp.Revoke, table, grant_config),
-        )
+        return self._dcl_grants_config_expr(exp.Revoke, table, grant_config)
 
     def _get_current_grants_config(self, table: exp.Table) -> GrantsConfig:
         """Returns current grants for a Postgres table as a dictionary."""

sqlmesh/core/model/meta.py

@@ -545,8 +545,7 @@ def parse_exp_to_str(e: exp.Expression) -> str:
                     if grantee:  # skip empty strings
                         grantee_list.append(grantee)
 
-                if grantee_list:
-                    grants_dict[permission_name.strip()] = grantee_list
+                grants_dict[permission_name.strip()] = grantee_list
 
         return grants_dict
 

sqlmesh/dbt/basemodel.py

@@ -153,7 +153,11 @@ def _validate_hooks(cls, v: t.Union[str, t.List[t.Union[SqlStr, str]]]) -> t.Lis
 
     @field_validator("grants", mode="before")
     @classmethod
-    def _validate_grants(cls, v: t.Dict[str, str]) -> t.Dict[str, t.List[str]]:
+    def _validate_grants(
+        cls, v: t.Optional[t.Dict[str, str]]
+    ) -> t.Optional[t.Dict[str, t.List[str]]]:
+        if v is None:
+            return None
         return {key: ensure_list(value) for key, value in v.items()}
 
     _FIELD_UPDATE_STRATEGY: t.ClassVar[t.Dict[str, UpdateStrategy]] = {

sqlmesh/dbt/model.py

@@ -627,6 +627,7 @@ def to_sqlmesh(
             if physical_properties:
                 model_kwargs["physical_properties"] = physical_properties
 
+        # A falsy grants config (None or {}) is considered as unmanaged per dbt semantics
         if self.grants:
             model_kwargs["grants"] = self.grants
 

tests/core/test_model.py

@@ -11690,6 +11690,13 @@ def test_grants_validation_no_grants():
     assert model.grants is None
 
 
+def test_grants_validation_empty_grantees():
+    model = create_sql_model(
+        "db.table", parse_one("SELECT 1 AS id"), kind="FULL", grants={"select": []}
+    )
+    assert model.grants == {"select": []}
+
+
 def test_grants_table_type_view():
     model = create_sql_model("test_view", parse_one("SELECT 1 as id"), kind="VIEW")
     assert model.grants_table_type == DataObjectType.VIEW

tests/core/test_snapshot_evaluator.py

@@ -4634,15 +4634,20 @@ def test_grants_unsupported_engine(make_mocked_engine_adapter, mocker):
     sync_grants_mock.assert_not_called()
 
 
-def test_grants_clears_grants(make_mocked_engine_adapter, mocker):
+def test_grants_revokes_permissions(make_mocked_engine_adapter, mocker):
     adapter = make_mocked_engine_adapter(EngineAdapter)
     adapter.SUPPORTS_GRANTS = True
     sync_grants_mock = mocker.patch.object(adapter, "_sync_grants_config")
     strategy = ViewStrategy(adapter)
-    model = create_sql_model("test_model", parse_one("SELECT 1 as id"), grants={})
+    model = create_sql_model("test_model", parse_one("SELECT 1 as id"), grants={"select": []})
+    model2 = create_sql_model("test_model2", parse_one("SELECT 1 as id"), grants={})
 
     strategy._apply_grants(model, "test_table", GrantsTargetLayer.PHYSICAL)
+    sync_grants_mock.assert_called_once()
+
+    sync_grants_mock.reset_mock()
 
+    strategy._apply_grants(model2, "test_table", GrantsTargetLayer.PHYSICAL)
     sync_grants_mock.assert_called_once()