add support to configure anonymous in s3 clients

pyiceberg/io/__init__.py

@@ -52,6 +52,7 @@
 AWS_SESSION_TOKEN = "client.session-token"
 AWS_ROLE_ARN = "client.role-arn"
 AWS_ROLE_SESSION_NAME = "client.role-session-name"
+S3_ANONYMOUS = "s3.anonymous"
 S3_ENDPOINT = "s3.endpoint"
 S3_ACCESS_KEY_ID = "s3.access-key-id"
 S3_SECRET_ACCESS_KEY = "s3.secret-access-key"

pyiceberg/io/fsspec.py

@@ -65,6 +65,7 @@
     HF_ENDPOINT,
     HF_TOKEN,
     S3_ACCESS_KEY_ID,
+    S3_ANONYMOUS,
     S3_CONNECT_TIMEOUT,
     S3_ENDPOINT,
     S3_PROXY_URI,
@@ -163,6 +164,9 @@ def _s3(properties: Properties) -> AbstractFileSystem:
 
     if request_timeout := properties.get(S3_REQUEST_TIMEOUT):
         config_kwargs["read_timeout"] = float(request_timeout)
+
+    if s3_anonymous := properties.get(S3_ANONYMOUS):
+        config_kwargs["anon"] = bool(s3_anonymous)
 
     fs = S3FileSystem(client_kwargs=client_kwargs, config_kwargs=config_kwargs)
 

tests/io/test_fsspec.py

@@ -265,6 +265,36 @@ def test_fsspec_s3_session_properties() -> None:
             config_kwargs={},
         )
 
+def test_fsspec_s3_session_properties_anon_config() -> None:
+    session_properties: Properties = {
+        "s3.anonymous": "true",
+        "s3.endpoint": "http://localhost:9000",
+        "s3.access-key-id": "admin",
+        "s3.secret-access-key": "password",
+        "s3.region": "us-east-1",
+        "s3.session-token": "s3.session-token",
+        **UNIFIED_AWS_SESSION_PROPERTIES,
+    }
+
+    with mock.patch("s3fs.S3FileSystem") as mock_s3fs:
+        s3_fileio = FsspecFileIO(properties=session_properties)
+        filename = str(uuid.uuid4())
+
+        s3_fileio.new_input(location=f"s3://warehouse/{filename}")
+
+        mock_s3fs.assert_called_with(
+            client_kwargs={
+                "endpoint_url": "http://localhost:9000",
+                "aws_access_key_id": "admin",
+                "aws_secret_access_key": "password",
+                "region_name": "us-east-1",
+                "aws_session_token": "s3.session-token",
+            },
+            config_kwargs={
+                "anon": True,
+            },
+        )
+
 
 def test_fsspec_unified_session_properties() -> None:
     session_properties: Properties = {