feat: Add AWS profile support to fsspec

Author: statsops

pyiceberg/io/fsspec.py

@@ -51,6 +51,7 @@
     ADLS_TENANT_ID,
     ADLS_TOKEN,
     AWS_ACCESS_KEY_ID,
+    AWS_PROFILE_NAME,
     AWS_REGION,
     AWS_SECRET_ACCESS_KEY,
     AWS_SESSION_TOKEN,
@@ -71,6 +72,7 @@
     S3_CONNECT_TIMEOUT,
     S3_ENDPOINT,
     S3_FORCE_VIRTUAL_ADDRESSING,
+    S3_PROFILE_NAME,
     S3_PROXY_URI,
     S3_REGION,
     S3_REQUEST_TIMEOUT,
@@ -205,7 +207,13 @@ def _s3(properties: Properties) -> AbstractFileSystem:
     else:
         anon = False
 
-    fs = S3FileSystem(anon=anon, client_kwargs=client_kwargs, config_kwargs=config_kwargs)
+    session = None
+    if profile_name := get_first_property_value(properties, S3_PROFILE_NAME, AWS_PROFILE_NAME):
+        from aiobotocore.session import AioSession
+
+        session = AioSession(profile=profile_name)
+
+    fs = S3FileSystem(anon=anon, client_kwargs=client_kwargs, config_kwargs=config_kwargs, session=session)
 
     for event_name, event_function in register_events.items():
         fs.s3.meta.events.unregister(event_name, unique_id=1925)

tests/io/test_fsspec_profile.py

@@ -0,0 +1,67 @@
+
+from unittest import mock
+import pytest
+from pyiceberg.io.fsspec import FsspecFileIO
+from pyiceberg.typedef import Properties
+
+UNIFIED_AWS_SESSION_PROPERTIES = {
+    "client.access-key-id": "client.access-key-id",
+    "client.secret-access-key": "client.secret-access-key",
+    "client.region": "client.region",
+    "client.session-token": "client.session-token",
+}
+import uuid
+
+def test_fsspec_s3_session_properties_with_profile() -> None:
+    session_properties: Properties = {
+        "s3.profile-name": "test-profile",
+        "s3.endpoint": "http://localhost:9000",
+        **UNIFIED_AWS_SESSION_PROPERTIES,
+    }
+
+    with mock.patch("s3fs.S3FileSystem") as mock_s3fs, mock.patch("aiobotocore.session.AioSession") as mock_aio_session:
+        s3_fileio = FsspecFileIO(properties=session_properties)
+        filename = str(uuid.uuid4())
+
+        s3_fileio.new_input(location=f"s3://warehouse/{filename}")
+
+        mock_aio_session.assert_called_with(profile="test-profile")
+        mock_s3fs.assert_called_with(
+            anon=False,
+            client_kwargs={
+                "endpoint_url": "http://localhost:9000",
+                "aws_access_key_id": "client.access-key-id",
+                "aws_secret_access_key": "client.secret-access-key",
+                "region_name": "client.region",
+                "aws_session_token": "client.session-token",
+            },
+            config_kwargs={},
+            session=mock_aio_session(),
+        )
+
+def test_fsspec_s3_session_properties_with_client_profile() -> None:
+    session_properties: Properties = {
+        "client.profile-name": "test-profile",
+        "s3.endpoint": "http://localhost:9000",
+        **UNIFIED_AWS_SESSION_PROPERTIES,
+    }
+
+    with mock.patch("s3fs.S3FileSystem") as mock_s3fs, mock.patch("aiobotocore.session.AioSession") as mock_aio_session:
+        s3_fileio = FsspecFileIO(properties=session_properties)
+        filename = str(uuid.uuid4())
+
+        s3_fileio.new_input(location=f"s3://warehouse/{filename}")
+
+        mock_aio_session.assert_called_with(profile="test-profile")
+        mock_s3fs.assert_called_with(
+            anon=False,
+            client_kwargs={
+                "endpoint_url": "http://localhost:9000",
+                "aws_access_key_id": "client.access-key-id",
+                "aws_secret_access_key": "client.secret-access-key",
+                "region_name": "client.region",
+                "aws_session_token": "client.session-token",
+            },
+            config_kwargs={},
+            session=mock_aio_session(),
+        )