You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ke...@apache.org on 2026/06/13 21:36:53 UTC
(iceberg-python) branch codex/load-credentials-table-integration created (now 8c4a69aae)
This is an automated email from the ASF dual-hosted git repository.
kevinjqliu pushed a change to branch codex/load-credentials-table-integration
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
at 8c4a69aae Load table credentials from credentials endpoint
This branch includes the following new commits:
new 8c4a69aae Load table credentials from credentials endpoint
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iceberg-python) 01/01: Load table credentials from credentials endpoint
Posted by ke...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kevinjqliu pushed a commit to branch codex/load-credentials-table-integration
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
commit 8c4a69aae33ad91acea9300e34fe052d3eae45fa
Author: Kevin Liu <ke...@gmail.com>
AuthorDate: Sat Jun 13 14:36:32 2026 -0700
Load table credentials from credentials endpoint
---
pyiceberg/catalog/rest/__init__.py | 24 +++++++++--
tests/catalog/test_rest.py | 86 ++++++++++++++++++++++++++++++++------
2 files changed, 93 insertions(+), 17 deletions(-)
diff --git a/pyiceberg/catalog/rest/__init__.py b/pyiceberg/catalog/rest/__init__.py
index c85ce0260..2cc0386ec 100644
--- a/pyiceberg/catalog/rest/__init__.py
+++ b/pyiceberg/catalog/rest/__init__.py
@@ -21,6 +21,7 @@ from enum import Enum
from typing import (
TYPE_CHECKING,
Any,
+ cast,
)
from urllib.parse import quote, unquote
@@ -233,6 +234,7 @@ class ScanPlanningMode(Enum):
ACCESS_DELEGATION_DEFAULT = "vended-credentials"
+ACCESS_DELEGATION_HEADER = "X-Iceberg-Access-Delegation"
AUTHORIZATION_HEADER = "Authorization"
BEARER_PREFIX = "Bearer"
CATALOG_SCOPE = "catalog"
@@ -486,6 +488,22 @@ class RestCatalog(Catalog):
return best_match.config if best_match else {}
+ def _should_load_credentials(self) -> bool:
+ access_delegation = cast(str, self._session.headers.get(ACCESS_DELEGATION_HEADER, ""))
+ return ACCESS_DELEGATION_DEFAULT in {delegation.strip().lower() for delegation in access_delegation.split(",")}
+
+ def _resolve_table_response_credentials(self, identifier_tuple: tuple[str, ...], table_response: TableResponse) -> Properties:
+ location = table_response.metadata_location or table_response.metadata.location
+ credential_config = self._resolve_storage_credentials(table_response.storage_credentials, location)
+ if table_response.storage_credentials or not location:
+ return credential_config
+
+ if Capability.V1_LOAD_CREDENTIALS not in self._supported_endpoints or not self._should_load_credentials():
+ return {}
+
+ credentials_response = self._load_credentials(identifier_tuple)
+ return self._resolve_storage_credentials(credentials_response.storage_credentials, location)
+
def _load_file_io(self, properties: Properties = EMPTY_DICT, location: str | None = None) -> FileIO:
merged_properties = {**self.properties, **properties}
if self._auth_manager:
@@ -853,9 +871,7 @@ class RestCatalog(Catalog):
def _response_to_table(self, identifier_tuple: tuple[str, ...], table_response: TableResponse) -> Table:
# Per Iceberg spec: storage-credentials take precedence over config
- credential_config = self._resolve_storage_credentials(
- table_response.storage_credentials, table_response.metadata_location
- )
+ credential_config = self._resolve_table_response_credentials(identifier_tuple, table_response)
return Table(
identifier=identifier_tuple,
metadata_location=table_response.metadata_location, # type: ignore
@@ -904,7 +920,7 @@ class RestCatalog(Catalog):
session.headers["Content-type"] = "application/json"
session.headers["User-Agent"] = f"PyIceberg/{__version__}"
session.headers["X-Client-Version"] = f"PyIceberg {__version__}"
- session.headers.setdefault("X-Iceberg-Access-Delegation", ACCESS_DELEGATION_DEFAULT)
+ session.headers.setdefault(ACCESS_DELEGATION_HEADER, ACCESS_DELEGATION_DEFAULT)
def _create_table(
self,
diff --git a/tests/catalog/test_rest.py b/tests/catalog/test_rest.py
index 23bb14cb5..9e745693b 100644
--- a/tests/catalog/test_rest.py
+++ b/tests/catalog/test_rest.py
@@ -104,7 +104,6 @@ TEST_SUPPORTED_ENDPOINTS = [
Capability.V1_DELETE_TABLE,
Capability.V1_RENAME_TABLE,
Capability.V1_REGISTER_TABLE,
- Capability.V1_LOAD_CREDENTIALS,
Capability.V1_LIST_VIEWS,
Capability.V1_LOAD_VIEW,
Capability.V1_VIEW_EXISTS,
@@ -162,17 +161,22 @@ def example_view_metadata_rest_json(example_view_metadata_v1: dict[str, Any]) ->
}
+def _mock_config_response(requests_mock: Mocker, endpoints: list[Endpoint] | None = None) -> None:
+ endpoints = TEST_SUPPORTED_ENDPOINTS if endpoints is None else endpoints
+ requests_mock.get(
+ f"{TEST_URI}v1/config",
+ json={"defaults": {}, "overrides": {}, "endpoints": [str(endpoint) for endpoint in endpoints]},
+ status_code=200,
+ )
+
+
@pytest.fixture
def rest_mock(requests_mock: Mocker) -> Mocker:
"""Takes the default requests_mock and adds the config endpoint to it
This endpoint is called when initializing the rest catalog
"""
- requests_mock.get(
- f"{TEST_URI}v1/config",
- json={"defaults": {}, "overrides": {}, "endpoints": [str(endpoint) for endpoint in TEST_SUPPORTED_ENDPOINTS]},
- status_code=200,
- )
+ _mock_config_response(requests_mock)
return requests_mock
@@ -1452,10 +1456,11 @@ def test_load_table_200_loading_mode(
def test_load_table_honor_access_delegation(
- rest_mock: Mocker, example_table_metadata_with_snapshot_v1_rest_json: dict[str, Any]
+ requests_mock: Mocker, example_table_metadata_with_snapshot_v1_rest_json: dict[str, Any]
) -> None:
+ _mock_config_response(requests_mock, [*TEST_SUPPORTED_ENDPOINTS, Capability.V1_LOAD_CREDENTIALS])
test_headers_with_remote_signing = {**TEST_HEADERS, "X-Iceberg-Access-Delegation": "remote-signing"}
- rest_mock.get(
+ requests_mock.get(
f"{TEST_URI}v1/namespaces/fokko/tables/table",
json=example_table_metadata_with_snapshot_v1_rest_json,
status_code=200,
@@ -3114,9 +3119,12 @@ def test_resolve_storage_credentials_empty() -> None:
assert RestCatalog._resolve_storage_credentials([], None) == {}
-def test_load_table_with_storage_credentials(rest_mock: Mocker, example_table_metadata_with_snapshot_v1: dict[str, Any]) -> None:
+def test_load_table_with_storage_credentials(
+ requests_mock: Mocker, example_table_metadata_with_snapshot_v1: dict[str, Any]
+) -> None:
+ _mock_config_response(requests_mock, [*TEST_SUPPORTED_ENDPOINTS, Capability.V1_LOAD_CREDENTIALS])
metadata_location = "s3://warehouse/database/table/metadata/00001.metadata.json"
- rest_mock.get(
+ requests_mock.get(
f"{TEST_URI}v1/namespaces/fokko/tables/table",
json={
"metadata-location": metadata_location,
@@ -3146,10 +3154,62 @@ def test_load_table_with_storage_credentials(rest_mock: Mocker, example_table_me
assert table.io.properties["s3.access-key-id"] == "vended-key"
assert table.io.properties["s3.secret-access-key"] == "vended-secret"
assert table.io.properties["s3.session-token"] == "vended-token"
+ assert len(requests_mock.request_history) == 2
-def test_load_credentials_with_longest_prefix(rest_mock: Mocker) -> None:
- rest_mock.get(
+def test_load_table_loads_credentials_when_endpoint_supported(
+ requests_mock: Mocker, example_table_metadata_with_snapshot_v1: dict[str, Any]
+) -> None:
+ _mock_config_response(requests_mock, [*TEST_SUPPORTED_ENDPOINTS, Capability.V1_LOAD_CREDENTIALS])
+ metadata_location = "s3://warehouse/database/table/metadata/00001.metadata.json"
+ requests_mock.get(
+ f"{TEST_URI}v1/namespaces/fokko/tables/table",
+ json={
+ "metadata-location": metadata_location,
+ "metadata": example_table_metadata_with_snapshot_v1,
+ "config": {
+ "s3.access-key-id": "from-config",
+ "s3.secret-access-key": "from-config-secret",
+ },
+ },
+ status_code=200,
+ request_headers=TEST_HEADERS,
+ )
+ requests_mock.get(
+ f"{TEST_URI}v1/namespaces/fokko/tables/table/credentials",
+ json={
+ "storage-credentials": [
+ {
+ "prefix": "s3://warehouse/database/",
+ "config": {"s3.access-key-id": "short-prefix-key"},
+ },
+ {
+ "prefix": "s3://warehouse/database/table",
+ "config": {
+ "s3.access-key-id": "long-prefix-key",
+ "s3.secret-access-key": "long-prefix-secret",
+ },
+ },
+ ],
+ },
+ status_code=200,
+ request_headers=TEST_HEADERS,
+ )
+ catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+ table = catalog.load_table(("fokko", "table"))
+
+ assert table.io.properties["s3.access-key-id"] == "long-prefix-key"
+ assert table.io.properties["s3.secret-access-key"] == "long-prefix-secret"
+ assert [request.url for request in requests_mock.request_history] == [
+ f"{TEST_URI}v1/config",
+ f"{TEST_URI}v1/namespaces/fokko/tables/table",
+ f"{TEST_URI}v1/namespaces/fokko/tables/table/credentials",
+ ]
+
+
+def test_load_credentials_with_longest_prefix(requests_mock: Mocker) -> None:
+ _mock_config_response(requests_mock, [*TEST_SUPPORTED_ENDPOINTS, Capability.V1_LOAD_CREDENTIALS])
+ requests_mock.get(
f"{TEST_URI}v1/namespaces/fokko/tables/table/credentials",
json={
"storage-credentials": [
@@ -3177,7 +3237,7 @@ def test_load_credentials_with_longest_prefix(rest_mock: Mocker) -> None:
)
assert credentials == {"s3.access-key-id": "long-prefix-key", "s3.secret-access-key": "long-prefix-secret"}
- assert rest_mock.last_request.url == f"{TEST_URI}v1/namespaces/fokko/tables/table/credentials"
+ assert requests_mock.last_request.url == f"{TEST_URI}v1/namespaces/fokko/tables/table/credentials"
def test_load_table_without_storage_credentials(