Fix async conn for none aws_session_token (#30868)

airflow/providers/amazon/aws/hooks/base_aws.py

@@ -138,9 +138,9 @@ def _apply_session_kwargs(self, session):
             or self.conn.session_kwargs.get("aws_session_token", None)
         ):
             session.set_credentials(
-                self.conn.session_kwargs["aws_access_key_id"],
-                self.conn.session_kwargs["aws_secret_access_key"],
-                self.conn.session_kwargs["aws_session_token"],
+                access_key=self.conn.session_kwargs.get("aws_access_key_id"),
+                secret_key=self.conn.session_kwargs.get("aws_secret_access_key"),
+                token=self.conn.session_kwargs.get("aws_session_token"),
             )
 
         if self.conn.session_kwargs.get("region_name", None) is not None:

tests/providers/amazon/aws/hooks/test_base_aws.py

@@ -246,6 +246,25 @@ def test_async_create_session_from_credentials(self, region_name, profile_name):
 
         assert session_profile == profile_name
 
+    @pytest.mark.asyncio
+    async def test_async_create_a_session_from_credentials_without_token(self):
+        mock_conn = Connection(
+            conn_type=MOCK_CONN_TYPE,
+            conn_id=MOCK_AWS_CONN_ID,
+            extra={
+                "aws_access_key_id": "test_aws_access_key_id",
+                "aws_secret_access_key": "test_aws_secret_access_key",
+                "region_name": "eu-central-1",
+            },
+        )
+        mock_conn_config = AwsConnectionWrapper(conn=mock_conn)
+        sf = BaseSessionFactory(conn=mock_conn_config, config=None)
+        async_session = sf.create_session(deferrable=True)
+        cred = await async_session.get_credentials()
+        assert cred.access_key == "test_aws_access_key_id"
+        assert cred.secret_key == "test_aws_secret_access_key"
+        assert cred.token is None
+
     config_for_credentials_test = [
         (
             "assume-with-initial-creds",