prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  Adding IAM parameters to redshift adapter
  ### Describe the feature
  I would the arguments to get_cluster_credentials to be added to the dbt profile configuration. In particular DbGroups to allow the temporary user to be added to a group and AutoCreate to allow auto creation of users that do not exist. 
  
  ### Describe alternatives you've considered
  Since these are IAM specific configurations the only other alternative is to not use the temporary credentials.
  
  ### Additional context
  This is a feature specifically for redshift users.
  
  ### Who will this benefit?
  This feature will be useful for dbt users who want to use temporary and dynamic credentials with redshift. 
  
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `plugins/redshift/dbt/adapters/redshift/connections.py`
  Content:
  ```
  1 from multiprocessing import Lock
  2 from contextlib import contextmanager
  3 from typing import NewType
  4 
  5 from dbt.adapters.postgres import PostgresConnectionManager
  6 from dbt.adapters.postgres import PostgresCredentials
  7 from dbt.logger import GLOBAL_LOGGER as logger  # noqa
  8 import dbt.exceptions
  9 import dbt.flags
  10 
  11 import boto3
  12 
  13 from hologram import FieldEncoder, JsonSchemaMixin
  14 from hologram.helpers import StrEnum
  15 
  16 from dataclasses import dataclass, field
  17 from typing import Optional
  18 
  19 drop_lock: Lock = dbt.flags.MP_CONTEXT.Lock()
  20 
  21 
  22 IAMDuration = NewType('IAMDuration', int)
  23 
  24 
  25 class IAMDurationEncoder(FieldEncoder):
  26     @property
  27     def json_schema(self):
  28         return {'type': 'integer', 'minimum': 0, 'maximum': 65535}
  29 
  30 
  31 JsonSchemaMixin.register_field_encoders({IAMDuration: IAMDurationEncoder()})
  32 
  33 
  34 class RedshiftConnectionMethod(StrEnum):
  35     DATABASE = 'database'
  36     IAM = 'iam'
  37 
  38 
  39 @dataclass
  40 class RedshiftCredentials(PostgresCredentials):
  41     method: RedshiftConnectionMethod = RedshiftConnectionMethod.DATABASE
  42     password: Optional[str] = None
  43     cluster_id: Optional[str] = field(
  44         default=None,
  45         metadata={'description': 'If using IAM auth, the name of the cluster'},
  46     )
  47     iam_duration_seconds: int = 900
  48     search_path: Optional[str] = None
  49     keepalives_idle: int = 240
  50 
  51     @property
  52     def type(self):
  53         return 'redshift'
  54 
  55     def _connection_keys(self):
  56         keys = super()._connection_keys()
  57         return keys + ('method', 'cluster_id', 'iam_duration_seconds')
  58 
  59 
  60 class RedshiftConnectionManager(PostgresConnectionManager):
  61     TYPE = 'redshift'
  62 
  63     @contextmanager
  64     def fresh_transaction(self, name=None):
  65         """On entrance to this context manager, hold an exclusive lock and
  66         create a fresh transaction for redshift, then commit and begin a new
  67         one before releasing the lock on exit.
  68 
  69         See drop_relation in RedshiftAdapter for more information.
  70 
  71         :param Optional[str] name: The name of the connection to use, or None
  72             to use the default.
  73         """
  74         with drop_lock:
  75             connection = self.get_thread_connection()
  76 
  77             if connection.transaction_open:
  78                 self.commit()
  79 
  80             self.begin()
  81             yield
  82 
  83             self.commit()
  84             self.begin()
  85 
  86     @classmethod
  87     def fetch_cluster_credentials(cls, db_user, db_name, cluster_id,
  88                                   duration_s):
  89         """Fetches temporary login credentials from AWS. The specified user
  90         must already exist in the database, or else an error will occur"""
  91         boto_client = boto3.client('redshift')
  92 
  93         try:
  94             return boto_client.get_cluster_credentials(
  95                 DbUser=db_user,
  96                 DbName=db_name,
  97                 ClusterIdentifier=cluster_id,
  98                 DurationSeconds=duration_s,
  99                 AutoCreate=False)
  100 
  101         except boto_client.exceptions.ClientError as e:
  102             raise dbt.exceptions.FailedToConnectException(
  103                 "Unable to get temporary Redshift cluster credentials: {}"
  104                 .format(e))
  105 
  106     @classmethod
  107     def get_tmp_iam_cluster_credentials(cls, credentials):
  108         cluster_id = credentials.cluster_id
  109 
  110         # default via:
  111         # boto3.readthedocs.io/en/latest/reference/services/redshift.html
  112         iam_duration_s = credentials.iam_duration_seconds
  113 
  114         if not cluster_id:
  115             raise dbt.exceptions.FailedToConnectException(
  116                 "'cluster_id' must be provided in profile if IAM "
  117                 "authentication method selected")
  118 
  119         cluster_creds = cls.fetch_cluster_credentials(
  120             credentials.user,
  121             credentials.database,
  122             credentials.cluster_id,
  123             iam_duration_s,
  124         )
  125 
  126         # replace username and password with temporary redshift credentials
  127         return credentials.replace(user=cluster_creds.get('DbUser'),
  128                                    password=cluster_creds.get('DbPassword'))
  129 
  130     @classmethod
  131     def get_credentials(cls, credentials):
  132         method = credentials.method
  133 
  134         # Support missing 'method' for backwards compatibility
  135         if method == 'database' or method is None:
  136             logger.debug("Connecting to Redshift using 'database' credentials")
  137             # this requirement is really annoying to encode into json schema,
  138             # so validate it here
  139             if credentials.password is None:
  140                 raise dbt.exceptions.FailedToConnectException(
  141                     "'password' field is required for 'database' credentials"
  142                 )
  143             return credentials
  144 
  145         elif method == 'iam':
  146             logger.debug("Connecting to Redshift using 'IAM' credentials")
  147             return cls.get_tmp_iam_cluster_credentials(credentials)
  148 
  149         else:
  150             raise dbt.exceptions.FailedToConnectException(
  151                 "Invalid 'method' in profile: '{}'".format(method))
  152
  ```
  --- END FILES ---
  
  Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.
  
  Here is an example:
  
  ```diff
  diff --git a/examples/server_async.py b/examples/server_async.py
  --- a/examples/server_async.py
  +++ b/examples/server_async.py
  @@ -313,4 +313,4 @@
   
   
   if __name__ == "__main__":
  -    asyncio.run(run_async_server("."), debug=True)
  +    asyncio.run(run_async_server(), debug=True)
  diff --git a/examples/server_sync.py b/examples/server_sync.py
  --- a/examples/server_sync.py
  +++ b/examples/server_sync.py
  @@ -313,5 +313,5 @@
   
   
   if __name__ == "__main__":
  -    server = run_sync_server(".")
  +    server = run_sync_server()
       server.shutdown()
  
  ```
  
golden_diff: diff --git a/plugins/redshift/dbt/adapters/redshift/connections.py b/plugins/redshift/dbt/adapters/redshift/connections.py
  --- a/plugins/redshift/dbt/adapters/redshift/connections.py
  +++ b/plugins/redshift/dbt/adapters/redshift/connections.py
  @@ -14,7 +14,7 @@
   from hologram.helpers import StrEnum
   
   from dataclasses import dataclass, field
  -from typing import Optional
  +from typing import Optional, List
   
   drop_lock: Lock = dbt.flags.MP_CONTEXT.Lock()
   
  @@ -47,6 +47,8 @@
       iam_duration_seconds: int = 900
       search_path: Optional[str] = None
       keepalives_idle: int = 240
  +    autocreate: bool = False
  +    db_groups: List[str] = field(default_factory=list)
   
       @property
       def type(self):
  @@ -85,7 +87,7 @@
   
       @classmethod
       def fetch_cluster_credentials(cls, db_user, db_name, cluster_id,
  -                                  duration_s):
  +                                  duration_s, autocreate, db_groups):
           """Fetches temporary login credentials from AWS. The specified user
           must already exist in the database, or else an error will occur"""
           boto_client = boto3.client('redshift')
  @@ -96,7 +98,8 @@
                   DbName=db_name,
                   ClusterIdentifier=cluster_id,
                   DurationSeconds=duration_s,
  -                AutoCreate=False)
  +                AutoCreate=autocreate,
  +                DbGroups=db_groups,)
   
           except boto_client.exceptions.ClientError as e:
               raise dbt.exceptions.FailedToConnectException(
  @@ -121,6 +124,8 @@
               credentials.database,
               credentials.cluster_id,
               iam_duration_s,
  +            credentials.autocreate,
  +            credentials.db_groups,
           )
   
           # replace username and password with temporary redshift credentials
  
verification_info: {"golden_diff": "diff --git a/plugins/redshift/dbt/adapters/redshift/connections.py b/plugins/redshift/dbt/adapters/redshift/connections.py\n--- a/plugins/redshift/dbt/adapters/redshift/connections.py\n+++ b/plugins/redshift/dbt/adapters/redshift/connections.py\n@@ -14,7 +14,7 @@\n from hologram.helpers import StrEnum\n \n from dataclasses import dataclass, field\n-from typing import Optional\n+from typing import Optional, List\n \n drop_lock: Lock = dbt.flags.MP_CONTEXT.Lock()\n \n@@ -47,6 +47,8 @@\n     iam_duration_seconds: int = 900\n     search_path: Optional[str] = None\n     keepalives_idle: int = 240\n+    autocreate: bool = False\n+    db_groups: List[str] = field(default_factory=list)\n \n     @property\n     def type(self):\n@@ -85,7 +87,7 @@\n \n     @classmethod\n     def fetch_cluster_credentials(cls, db_user, db_name, cluster_id,\n-                                  duration_s):\n+                                  duration_s, autocreate, db_groups):\n         \"\"\"Fetches temporary login credentials from AWS. The specified user\n         must already exist in the database, or else an error will occur\"\"\"\n         boto_client = boto3.client('redshift')\n@@ -96,7 +98,8 @@\n                 DbName=db_name,\n                 ClusterIdentifier=cluster_id,\n                 DurationSeconds=duration_s,\n-                AutoCreate=False)\n+                AutoCreate=autocreate,\n+                DbGroups=db_groups,)\n \n         except boto_client.exceptions.ClientError as e:\n             raise dbt.exceptions.FailedToConnectException(\n@@ -121,6 +124,8 @@\n             credentials.database,\n             credentials.cluster_id,\n             iam_duration_s,\n+            credentials.autocreate,\n+            credentials.db_groups,\n         )\n \n         # replace username and password with temporary redshift credentials\n", "issue": "Adding IAM parameters to redshift adapter\n### Describe the feature\r\nI would the arguments to get_cluster_credentials to be added to the dbt profile configuration. In particular DbGroups to allow the temporary user to be added to a group and AutoCreate to allow auto creation of users that do not exist. \r\n\r\n### Describe alternatives you've considered\r\nSince these are IAM specific configurations the only other alternative is to not use the temporary credentials.\r\n\r\n### Additional context\r\nThis is a feature specifically for redshift users.\r\n\r\n### Who will this benefit?\r\nThis feature will be useful for dbt users who want to use temporary and dynamic credentials with redshift. \r\n\n", "before_files": [{"content": "from multiprocessing import Lock\nfrom contextlib import contextmanager\nfrom typing import NewType\n\nfrom dbt.adapters.postgres import PostgresConnectionManager\nfrom dbt.adapters.postgres import PostgresCredentials\nfrom dbt.logger import GLOBAL_LOGGER as logger  # noqa\nimport dbt.exceptions\nimport dbt.flags\n\nimport boto3\n\nfrom hologram import FieldEncoder, JsonSchemaMixin\nfrom hologram.helpers import StrEnum\n\nfrom dataclasses import dataclass, field\nfrom typing import Optional\n\ndrop_lock: Lock = dbt.flags.MP_CONTEXT.Lock()\n\n\nIAMDuration = NewType('IAMDuration', int)\n\n\nclass IAMDurationEncoder(FieldEncoder):\n    @property\n    def json_schema(self):\n        return {'type': 'integer', 'minimum': 0, 'maximum': 65535}\n\n\nJsonSchemaMixin.register_field_encoders({IAMDuration: IAMDurationEncoder()})\n\n\nclass RedshiftConnectionMethod(StrEnum):\n    DATABASE = 'database'\n    IAM = 'iam'\n\n\n@dataclass\nclass RedshiftCredentials(PostgresCredentials):\n    method: RedshiftConnectionMethod = RedshiftConnectionMethod.DATABASE\n    password: Optional[str] = None\n    cluster_id: Optional[str] = field(\n        default=None,\n        metadata={'description': 'If using IAM auth, the name of the cluster'},\n    )\n    iam_duration_seconds: int = 900\n    search_path: Optional[str] = None\n    keepalives_idle: int = 240\n\n    @property\n    def type(self):\n        return 'redshift'\n\n    def _connection_keys(self):\n        keys = super()._connection_keys()\n        return keys + ('method', 'cluster_id', 'iam_duration_seconds')\n\n\nclass RedshiftConnectionManager(PostgresConnectionManager):\n    TYPE = 'redshift'\n\n    @contextmanager\n    def fresh_transaction(self, name=None):\n        \"\"\"On entrance to this context manager, hold an exclusive lock and\n        create a fresh transaction for redshift, then commit and begin a new\n        one before releasing the lock on exit.\n\n        See drop_relation in RedshiftAdapter for more information.\n\n        :param Optional[str] name: The name of the connection to use, or None\n            to use the default.\n        \"\"\"\n        with drop_lock:\n            connection = self.get_thread_connection()\n\n            if connection.transaction_open:\n                self.commit()\n\n            self.begin()\n            yield\n\n            self.commit()\n            self.begin()\n\n    @classmethod\n    def fetch_cluster_credentials(cls, db_user, db_name, cluster_id,\n                                  duration_s):\n        \"\"\"Fetches temporary login credentials from AWS. The specified user\n        must already exist in the database, or else an error will occur\"\"\"\n        boto_client = boto3.client('redshift')\n\n        try:\n            return boto_client.get_cluster_credentials(\n                DbUser=db_user,\n                DbName=db_name,\n                ClusterIdentifier=cluster_id,\n                DurationSeconds=duration_s,\n                AutoCreate=False)\n\n        except boto_client.exceptions.ClientError as e:\n            raise dbt.exceptions.FailedToConnectException(\n                \"Unable to get temporary Redshift cluster credentials: {}\"\n                .format(e))\n\n    @classmethod\n    def get_tmp_iam_cluster_credentials(cls, credentials):\n        cluster_id = credentials.cluster_id\n\n        # default via:\n        # boto3.readthedocs.io/en/latest/reference/services/redshift.html\n        iam_duration_s = credentials.iam_duration_seconds\n\n        if not cluster_id:\n            raise dbt.exceptions.FailedToConnectException(\n                \"'cluster_id' must be provided in profile if IAM \"\n                \"authentication method selected\")\n\n        cluster_creds = cls.fetch_cluster_credentials(\n            credentials.user,\n            credentials.database,\n            credentials.cluster_id,\n            iam_duration_s,\n        )\n\n        # replace username and password with temporary redshift credentials\n        return credentials.replace(user=cluster_creds.get('DbUser'),\n                                   password=cluster_creds.get('DbPassword'))\n\n    @classmethod\n    def get_credentials(cls, credentials):\n        method = credentials.method\n\n        # Support missing 'method' for backwards compatibility\n        if method == 'database' or method is None:\n            logger.debug(\"Connecting to Redshift using 'database' credentials\")\n            # this requirement is really annoying to encode into json schema,\n            # so validate it here\n            if credentials.password is None:\n                raise dbt.exceptions.FailedToConnectException(\n                    \"'password' field is required for 'database' credentials\"\n                )\n            return credentials\n\n        elif method == 'iam':\n            logger.debug(\"Connecting to Redshift using 'IAM' credentials\")\n            return cls.get_tmp_iam_cluster_credentials(credentials)\n\n        else:\n            raise dbt.exceptions.FailedToConnectException(\n                \"Invalid 'method' in profile: '{}'\".format(method))\n", "path": "plugins/redshift/dbt/adapters/redshift/connections.py"}], "after_files": [{"content": "from multiprocessing import Lock\nfrom contextlib import contextmanager\nfrom typing import NewType\n\nfrom dbt.adapters.postgres import PostgresConnectionManager\nfrom dbt.adapters.postgres import PostgresCredentials\nfrom dbt.logger import GLOBAL_LOGGER as logger  # noqa\nimport dbt.exceptions\nimport dbt.flags\n\nimport boto3\n\nfrom hologram import FieldEncoder, JsonSchemaMixin\nfrom hologram.helpers import StrEnum\n\nfrom dataclasses import dataclass, field\nfrom typing import Optional, List\n\ndrop_lock: Lock = dbt.flags.MP_CONTEXT.Lock()\n\n\nIAMDuration = NewType('IAMDuration', int)\n\n\nclass IAMDurationEncoder(FieldEncoder):\n    @property\n    def json_schema(self):\n        return {'type': 'integer', 'minimum': 0, 'maximum': 65535}\n\n\nJsonSchemaMixin.register_field_encoders({IAMDuration: IAMDurationEncoder()})\n\n\nclass RedshiftConnectionMethod(StrEnum):\n    DATABASE = 'database'\n    IAM = 'iam'\n\n\n@dataclass\nclass RedshiftCredentials(PostgresCredentials):\n    method: RedshiftConnectionMethod = RedshiftConnectionMethod.DATABASE\n    password: Optional[str] = None\n    cluster_id: Optional[str] = field(\n        default=None,\n        metadata={'description': 'If using IAM auth, the name of the cluster'},\n    )\n    iam_duration_seconds: int = 900\n    search_path: Optional[str] = None\n    keepalives_idle: int = 240\n    autocreate: bool = False\n    db_groups: List[str] = field(default_factory=list)\n\n    @property\n    def type(self):\n        return 'redshift'\n\n    def _connection_keys(self):\n        keys = super()._connection_keys()\n        return keys + ('method', 'cluster_id', 'iam_duration_seconds')\n\n\nclass RedshiftConnectionManager(PostgresConnectionManager):\n    TYPE = 'redshift'\n\n    @contextmanager\n    def fresh_transaction(self, name=None):\n        \"\"\"On entrance to this context manager, hold an exclusive lock and\n        create a fresh transaction for redshift, then commit and begin a new\n        one before releasing the lock on exit.\n\n        See drop_relation in RedshiftAdapter for more information.\n\n        :param Optional[str] name: The name of the connection to use, or None\n            to use the default.\n        \"\"\"\n        with drop_lock:\n            connection = self.get_thread_connection()\n\n            if connection.transaction_open:\n                self.commit()\n\n            self.begin()\n            yield\n\n            self.commit()\n            self.begin()\n\n    @classmethod\n    def fetch_cluster_credentials(cls, db_user, db_name, cluster_id,\n                                  duration_s, autocreate, db_groups):\n        \"\"\"Fetches temporary login credentials from AWS. The specified user\n        must already exist in the database, or else an error will occur\"\"\"\n        boto_client = boto3.client('redshift')\n\n        try:\n            return boto_client.get_cluster_credentials(\n                DbUser=db_user,\n                DbName=db_name,\n                ClusterIdentifier=cluster_id,\n                DurationSeconds=duration_s,\n                AutoCreate=autocreate,\n                DbGroups=db_groups,)\n\n        except boto_client.exceptions.ClientError as e:\n            raise dbt.exceptions.FailedToConnectException(\n                \"Unable to get temporary Redshift cluster credentials: {}\"\n                .format(e))\n\n    @classmethod\n    def get_tmp_iam_cluster_credentials(cls, credentials):\n        cluster_id = credentials.cluster_id\n\n        # default via:\n        # boto3.readthedocs.io/en/latest/reference/services/redshift.html\n        iam_duration_s = credentials.iam_duration_seconds\n\n        if not cluster_id:\n            raise dbt.exceptions.FailedToConnectException(\n                \"'cluster_id' must be provided in profile if IAM \"\n                \"authentication method selected\")\n\n        cluster_creds = cls.fetch_cluster_credentials(\n            credentials.user,\n            credentials.database,\n            credentials.cluster_id,\n            iam_duration_s,\n            credentials.autocreate,\n            credentials.db_groups,\n        )\n\n        # replace username and password with temporary redshift credentials\n        return credentials.replace(user=cluster_creds.get('DbUser'),\n                                   password=cluster_creds.get('DbPassword'))\n\n    @classmethod\n    def get_credentials(cls, credentials):\n        method = credentials.method\n\n        # Support missing 'method' for backwards compatibility\n        if method == 'database' or method is None:\n            logger.debug(\"Connecting to Redshift using 'database' credentials\")\n            # this requirement is really annoying to encode into json schema,\n            # so validate it here\n            if credentials.password is None:\n                raise dbt.exceptions.FailedToConnectException(\n                    \"'password' field is required for 'database' credentials\"\n                )\n            return credentials\n\n        elif method == 'iam':\n            logger.debug(\"Connecting to Redshift using 'IAM' credentials\")\n            return cls.get_tmp_iam_cluster_credentials(credentials)\n\n        else:\n            raise dbt.exceptions.FailedToConnectException(\n                \"Invalid 'method' in profile: '{}'\".format(method))\n", "path": "plugins/redshift/dbt/adapters/redshift/connections.py"}]}