Check storage-credentials before config in table response

Author: Alexandra Popova

pyiceberg/catalog/rest/__init__.py

@@ -156,6 +156,7 @@ 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)
 
 
 class CreateTableRequest(IcebergBaseModel):
@@ -454,7 +455,11 @@ def _response_to_table(self, identifier_tuple: Tuple[str, ...], table_response:
             metadata_location=table_response.metadata_location,  # type: ignore
             metadata=table_response.metadata,
             io=self._load_file_io(
-                {**table_response.metadata.properties, **table_response.config}, table_response.metadata_location
+                {
+                    **table_response.metadata.properties,
+                    **(table_response.storage_credentials or table_response.config),
+                },
+                table_response.metadata_location,
             ),
             catalog=self,
             config=table_response.config,
@@ -466,7 +471,11 @@ def _response_to_staged_table(self, identifier_tuple: Tuple[str, ...], table_res
             metadata_location=table_response.metadata_location,  # type: ignore
             metadata=table_response.metadata,
             io=self._load_file_io(
-                {**table_response.metadata.properties, **table_response.config}, table_response.metadata_location
+                {
+                    **table_response.metadata.properties,
+                    **(table_response.storage_credentials or table_response.config),
+                },
+                table_response.metadata_location,
             ),
             catalog=self,
         )

tests/catalog/test_rest.py

@@ -18,6 +18,7 @@
 import base64
 import os
 from typing import Any, Callable, Dict, cast
+from copy import deepcopy
 from unittest import mock
 
 import pytest
@@ -858,6 +859,42 @@ def test_load_table_200(rest_mock: Mocker, example_table_metadata_with_snapshot_
     assert actual == expected
 
 
+def test_load_table_prefers_storage_credentials_over_config(
+    rest_mock: Mocker, example_table_metadata_with_snapshot_v1_rest_json: Dict[str, Any], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    table_resp = deepcopy(example_table_metadata_with_snapshot_v1_rest_json)
+    table_resp["config"] = {"some.key": "from-config", "only.config": "only-config"}
+    table_resp["storage-credentials"] = {"some.key": "from-cred", "only.creds": "only-creds"}
+
+    table_resp["metadata"].setdefault("properties", {})["meta.key"] = "from-metadata"
+
+    captured: Dict[str, Any] = {}
+
+    def _capture_io(_self: Any, properties: Dict[str, Any], location: Any) -> Any:  # type: ignore
+        captured["properties"] = dict(properties)
+        captured["location"] = location
+        class _DummyIO:
+            pass
+        return _DummyIO()
+
+    monkeypatch.setattr(RestCatalog, "_load_file_io", _capture_io, raising=True)
+
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces/fokko/tables/table",
+        json=table_resp,
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    _ = catalog.load_table(("fokko", "table"))
+
+    props = cast(Dict[str, Any], captured["properties"])
+    assert props["meta.key"] == "from-metadata"
+    assert props["some.key"] == "from-cred"
+    assert props["only.creds"] == "only-creds"
+
+
 def test_load_table_200_loading_mode(
     rest_mock: Mocker, example_table_metadata_with_snapshot_v1_rest_json: Dict[str, Any]
 ) -> None: