Bug: Passing Session to Connection, doesn't update client args for S3fs

[aaronclong]

Currently, when a session is passed, the client args aren't updated to reflect the session's credentials.
This is important for the PandasCursor because it passes the client args to s3fs to grab the files. It can create access issues and get failures.

• PyAthena/pyathena/connection.py

  Lines 98 to 99 in 0f6b44d

  	if session:
  	self._session = session

• PyAthena/pyathena/connection.py

  Lines 35 to 37 in 0f6b44d

  	"aws_access_key_id",
  	"aws_secret_access_key",
  	"aws_session_token",

• PyAthena/pyathena/connection.py

  Lines 209 to 211 in 0f6b44d

  	@property
  	def _client_kwargs(self) -> Dict[str, Any]:
  	return {k: v for k, v in self._kwargs.items() if k in self._CLIENT_PASSING_ARGS}

• PyAthena/pyathena/pandas/result_set.py

  Line 174 in 0f6b44d

  **self.connection._client_kwargs,

[aaronclong]

A possible work around is to set these credentials as kwargs for now, but PyAthena could also handle this in the future.

if session: 
     self._session = session 
     creds = session.get_credentials()
     cred_value = creds.get_frozen_credentials()
     self._kwargs.update(
           {
               "aws_access_key_id": cred_value.access_key,
                "aws_secret_access_key": cred_value.secret_key,
                "aws_session_token": cred_value.token
             }
      )

[laughingman7743]

Thank you. The workaround implementation looks good, but it would be nice if the Session object could be passed to s3fs.
https://github.com/fsspec/s3fs/blob/main/s3fs/core.py#L271
It seems to be expected that the AioSession object can be passed though.
https://github.com/aio-libs/aiobotocore/blob/master/aiobotocore/session.py#L33

[laughingman7743]

It is difficult to convert a botocore Session object to an AioSession.
I have tried the following fix, but I am not sure if this fix is correct for me.
#378

    def __s3_file_system(self):
        from s3fs import S3FileSystem

        creds = self.connection.session.get_credentials().get_frozen_credentials()
        client_kwargs = self.connection._client_kwargs
        client_kwargs.update(
            {
                "aws_access_key_id": creds.access_key,
                "aws_secret_access_key": creds.secret_key,
                "aws_session_token": creds.token,
            }
        )
        return S3FileSystem(
            profile=self.connection.profile_name,
            client_kwargs={
                "region_name": self.connection.region_name,
                **client_kwargs,
            },
        )