Repository: dbt-labs/dbt-redshift (python)
Issue: [Bug] dbt-redshift should not use `group_federation=true` when connecting with Serverless.
Group_federation parameter is not required with Serverless and makes the redshift connector call `getclustercredentialswithIAM` instead of `get_credentials` - the former doesn't exist for Serverless.
Pull request description: Remove group_federation for serverless
See this convo: https://dbt-labs.slack.com/archives/C04FN4114AD/p1717037952388479

Resolves #835 835

This has no test :/
But I confirmed the connection now hits redshift instead or returning:
```
  >Database Error
  'RedshiftServerless' object has no attribute 'get_cluster_credentials_with_iam'
```
 

Patch:
diff --git a/.changes/unreleased/Fixes-20240531-113620.yaml b/.changes/unreleased/Fixes-20240531-113620.yaml
new file mode 100644
index 00000000..a020ca04
--- /dev/null
+++ b/.changes/unreleased/Fixes-20240531-113620.yaml
@@ -0,0 +1,6 @@
+kind: Fixes
+body: Support IAM Role authentication for Redshift Serverless
+time: 2024-05-31T11:36:20.397521-07:00
+custom:
+  Author: fleid
+  Issue: "835"
diff --git a/dbt/adapters/redshift/connections.py b/dbt/adapters/redshift/connections.py
index dfd7ab30..d3fbcafe 100644
--- a/dbt/adapters/redshift/connections.py
+++ b/dbt/adapters/redshift/connections.py
@@ -237,10 +237,15 @@ class RedshiftConnectMethodFactory:
     def _iam_role_kwargs(self) -> Dict[str, Optional[Any]]:
         logger.debug("Connecting to redshift with 'iam_role' credentials method")
         kwargs = self._iam_kwargs
-        kwargs.update(
-            group_federation=True,
-            db_user=None,
-        )
+
+        # It's a role, we're ignoring the user
+        kwargs.update(db_user=None)
+
+        # Serverless shouldn't get group_federation, Provisoned clusters should
+        if "serverless" in self.credentials.host:
+            kwargs.update(group_federation=False)
+        else:
+            kwargs.update(group_federation=True)
 
         if iam_profile := self.credentials.iam_profile:
             kwargs.update(profile=iam_profile)
@@ -256,10 +261,10 @@ class RedshiftConnectMethodFactory:
             password="",
         )
 
-        if cluster_id := self.credentials.cluster_id:
-            kwargs.update(cluster_identifier=cluster_id)
-        elif "serverless" in self.credentials.host:
+        if "serverless" in self.credentials.host:
             kwargs.update(cluster_identifier=None)
+        elif cluster_id := self.credentials.cluster_id:
+            kwargs.update(cluster_identifier=cluster_id)
         else:
             raise FailedToConnectError(
                 "Failed to use IAM method:"


Test patch:
diff --git a/tests/unit/test_auth_method.py b/tests/unit/test_auth_method.py
index bd9912d0..55b1aad7 100644
--- a/tests/unit/test_auth_method.py
+++ b/tests/unit/test_auth_method.py
@@ -456,3 +456,120 @@ class TestIAMRoleMethod(AuthMethod):
             group_federation=True,
             **DEFAULT_SSL_CONFIG,
         )
+
+
+class TestIAMRoleMethodServerless(AuthMethod):
+    # Should behave like IAM Role provisioned, with the exception of not having group_federation set
+
+    @mock.patch("redshift_connector.connect", MagicMock())
+    def test_profile_default_region(self):
+        self.config.credentials = self.config.credentials.replace(
+            method="iam_role",
+            iam_profile="iam_profile_test",
+            host="doesnotexist.1233.us-east-2.redshift-serverless.amazonaws.com",
+        )
+        connection = self.adapter.acquire_connection("dummy")
+        connection.handle
+        redshift_connector.connect.assert_called_once_with(
+            iam=True,
+            host="doesnotexist.1233.us-east-2.redshift-serverless.amazonaws.com",
+            database="redshift",
+            cluster_identifier=None,
+            region=None,
+            auto_create=False,
+            db_groups=[],
+            db_user=None,
+            password="",
+            user="",
+            profile="iam_profile_test",
+            timeout=None,
+            port=5439,
+            group_federation=False,
+            **DEFAULT_SSL_CONFIG,
+        )
+
+    @mock.patch("redshift_connector.connect", MagicMock())
+    def test_profile_ignore_cluster(self):
+        self.config.credentials = self.config.credentials.replace(
+            method="iam_role",
+            iam_profile="iam_profile_test",
+            host="doesnotexist.1233.us-east-2.redshift-serverless.amazonaws.com",
+            cluster_id="my_redshift",
+        )
+        connection = self.adapter.acquire_connection("dummy")
+        connection.handle
+        redshift_connector.connect.assert_called_once_with(
+            iam=True,
+            host="doesnotexist.1233.us-east-2.redshift-serverless.amazonaws.com",
+            database="redshift",
+            cluster_identifier=None,
+            region=None,
+            auto_create=False,
+            db_groups=[],
+            db_user=None,
+            password="",
+            user="",
+            profile="iam_profile_test",
+            timeout=None,
+            port=5439,
+            group_federation=False,
+            **DEFAULT_SSL_CONFIG,
+        )
+
+    @mock.patch("redshift_connector.connect", MagicMock())
+    def test_profile_explicit_region(self):
+        # Successful test
+        self.config.credentials = self.config.credentials.replace(
+            method="iam_role",
+            iam_profile="iam_profile_test",
+            host="doesnotexist.1233.redshift-serverless.amazonaws.com",
+            region="us-east-2",
+        )
+        connection = self.adapter.acquire_connection("dummy")
+        connection.handle
+        redshift_connector.connect.assert_called_once_with(
+            iam=True,
+            host="doesnotexist.1233.redshift-serverless.amazonaws.com",
+            database="redshift",
+            cluster_identifier=None,
+            region="us-east-2",
+            auto_create=False,
+            db_groups=[],
+            db_user=None,
+            password="",
+            user="",
+            profile="iam_profile_test",
+            timeout=None,
+            port=5439,
+            group_federation=False,
+            **DEFAULT_SSL_CONFIG,
+        )
+
+    @mock.patch("redshift_connector.connect", MagicMock())
+    def test_profile_invalid_serverless(self):
+        self.config.credentials = self.config.credentials.replace(
+            method="iam_role",
+            iam_profile="iam_profile_test",
+            host="doesnotexist.1233.us-east-2.redshift-srvrlss.amazonaws.com",
+        )
+        with self.assertRaises(FailedToConnectError) as context:
+            connection = self.adapter.acquire_connection("dummy")
+            connection.handle
+            redshift_connector.connect.assert_called_once_with(
+                iam=True,
+                host="doesnotexist.1233.us-east-2.redshift-srvrlss.amazonaws.com",
+                database="redshift",
+                cluster_identifier=None,
+                region=None,
+                auto_create=False,
+                db_groups=[],
+                db_user=None,
+                password="",
+                user="",
+                profile="iam_profile_test",
+                port=5439,
+                timeout=None,
+                group_federation=False,
+                **DEFAULT_SSL_CONFIG,
+            )
+        self.assertTrue("'host' must be provided" in context.exception.msg)