fix storage credentials type

Author: Alexandra Popova

pyiceberg/catalog/rest/__init__.py

@@ -152,11 +152,16 @@ def _retry_hook(retry_state: RetryCallState) -> None:
 }
 
 
+class StorageCredential(IcebergBaseModel):
+    prefix: str = Field()
+    config: Properties = Field()
+
+
 class TableResponse(IcebergBaseModel):
     metadata_location: Optional[str] = Field(alias="metadata-location", default=None)
     metadata: TableMetadata
     config: Properties = Field(default_factory=dict)
-    storage_credentials: Optional[Properties] = Field(alias="storage-credentials", default=None)
+    storage_credentials: Optional[List[StorageCredential]] = Field(alias="storage-credentials", default=None)
 
 
 class CreateTableRequest(IcebergBaseModel):
@@ -457,7 +462,7 @@ def _response_to_table(self, identifier_tuple: Tuple[str, ...], table_response:
             io=self._load_file_io(
                 {
                     **table_response.metadata.properties,
-                    **(table_response.storage_credentials or table_response.config),
+                    **self._get_credentials(table_response.storage_credentials, table_response.config),
                 },
                 table_response.metadata_location,
             ),
@@ -473,13 +478,27 @@ def _response_to_staged_table(self, identifier_tuple: Tuple[str, ...], table_res
             io=self._load_file_io(
                 {
                     **table_response.metadata.properties,
-                    **(table_response.storage_credentials or table_response.config),
+                    **self._get_credentials(table_response.storage_credentials, table_response.config),
                 },
                 table_response.metadata_location,
             ),
             catalog=self,
         )
 
+    def _get_credentials(self, storage_credentials: List[StorageCredential], config: Properties) -> Properties:
+        if storage_credentials:
+            return self._get_storage_credentials(storage_credentials)
+        else:
+            return config
+
+    @staticmethod
+    def _get_storage_credentials(storage_credentials: List[StorageCredential]) -> Properties:
+        credentials: List[StorageCredential] = [sc for sc in storage_credentials if sc.prefix.startswith("s3")]
+        if len(credentials) > 1:
+            raise ValueError("Multiple S3 storage credentials found")
+        elif len(credentials) == 1:
+            return credentials[0].config
+
     def _refresh_token(self) -> None:
         # Reactive token refresh is atypical - we should proactively refresh tokens in a separate thread
         # instead of retrying on Auth Exceptions. Keeping refresh behavior for the LegacyOAuth2AuthManager

tests/catalog/test_rest.py

@@ -865,9 +865,14 @@ def test_storage_credentials_over_config(rest_mock: Mocker, example_table_metada
         "config": {
             AWS_ACCESS_KEY_ID: "from_config",
         },
-        "storage-credentials": {
-            AWS_ACCESS_KEY_ID: "from_storage_credentials",
-        },
+        "storage-credentials": [
+            {
+                "prefix": "s3://warehouse/database/",
+                "config": {
+                    AWS_ACCESS_KEY_ID: "from_storage_credentials",
+                },
+            }
+        ],
     }
 
     rest_mock.get(