class RestCatalog(Catalog):
uri: str
_session: Session
_auth_manager: AuthManager | None
_supported_endpoints: set[Endpoint]
_namespace_separator: str
def __init__(self, name: str, **properties: str):
"""Rest Catalog.
You either need to provide a client_id and client_secret, or an already valid token.
Args:
name: Name to identify the catalog.
properties: Properties that are passed along to the configuration.
"""
super().__init__(name, **properties)
self._auth_manager: AuthManager | None = None
self.uri = properties[URI]
self._fetch_config()
self._session = self._create_session()
def _create_session(self) -> Session:
"""Create a request session with provided catalog configuration."""
session = Session()
# Mount the retry/timeout adapter when `connection.*` properties are set.
# SigV4's adapter mounted below at `self.uri` is a longer prefix and still wins for that host.
if (connection_adapter := _create_connection_adapter(self.properties)) is not None:
session.mount("http://", connection_adapter)
session.mount("https://", connection_adapter)
# Set HTTP headers
self._config_headers(session)
# Sets the client side and server side SSL cert verification, if provided as properties.
if ssl_config := self.properties.get(SSL):
if (ssl_ca_bundle := ssl_config.get(CA_BUNDLE)) is not None:
session.verify = ssl_ca_bundle
if ssl_client := ssl_config.get(CLIENT):
if all(k in ssl_client for k in (CERT, KEY)):
session.cert = (ssl_client[CERT], ssl_client[KEY])
elif ssl_client_cert := ssl_client.get(CERT):
session.cert = ssl_client_cert
if auth_config := self.properties.get(AUTH):
auth_type = auth_config.get("type")
if auth_type is None:
raise ValueError("auth.type must be defined")
auth_type_config = auth_config.get(auth_type, {})
auth_impl = auth_config.get("impl")
if auth_type == CUSTOM and not auth_impl:
raise ValueError("auth.impl must be specified when using custom auth.type")
if auth_type != CUSTOM and auth_impl:
raise ValueError("auth.impl can only be specified when using custom auth.type")
self._auth_manager = AuthManagerFactory.create(auth_impl or auth_type, auth_type_config)
session.auth = AuthManagerAdapter(self._auth_manager)
else:
self._auth_manager = self._create_legacy_oauth2_auth_manager(session)
session.auth = AuthManagerAdapter(self._auth_manager)
# Configure SigV4 Request Signing
if property_as_bool(self.properties, SIGV4, False):
self._init_sigv4(session)
return session
@staticmethod
def _resolve_storage_credentials(storage_credentials: list[StorageCredential], location: str | None) -> Properties:
"""Resolve the best-matching storage credential by longest prefix match.
Mirrors the Java implementation in S3FileIO.clientForStoragePath() which iterates
over storage credential prefixes and selects the one with the longest match.
See: https://github.com/apache/iceberg/blob/main/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java
"""
if not storage_credentials or not location:
return {}
best_match: StorageCredential | None = None
for cred in storage_credentials:
if location.startswith(cred.prefix):
if best_match is None or len(cred.prefix) > len(best_match.prefix):
best_match = cred
return best_match.config if best_match else {}
def _load_file_io(self, properties: Properties = EMPTY_DICT, location: str | None = None) -> FileIO:
merged_properties = {**self.properties, **properties}
if self._auth_manager:
merged_properties[AUTH_MANAGER] = self._auth_manager
return load_file_io(merged_properties, location)
def _effective_scan_planning_mode(self, table_config: Properties) -> ScanPlanningMode:
"""Resolve the scan planning mode, where a loadTable override wins over the catalog property.
An invalid catalog-level value is ignored (with a warning) so it cannot block a valid
loadTable override or the default client-side mode. An invalid loadTable value still fails.
"""
# Parse the table override first so a valid loadTable value is not blocked by a bad catalog property.
table_mode = _parse_scan_planning_mode(table_config)
catalog_mode = _parse_scan_planning_mode(self.properties, strict=False)
if catalog_mode is not None and table_mode is not None and catalog_mode != table_mode:
logger.warning(
"Scan planning mode mismatch: client config=%s, server config=%s. Server config takes precedence.",
catalog_mode.value,
table_mode.value,
)
return table_mode or catalog_mode or ScanPlanningMode(SCAN_PLANNING_MODE_DEFAULT)
@override
def supports_server_side_planning(self, table_config: Properties = EMPTY_DICT) -> bool:
"""Check if server-side scan planning should be used, honoring a per-table loadTable override."""
if Capability.V1_SUBMIT_TABLE_SCAN_PLAN not in self._supported_endpoints:
return False
return self._effective_scan_planning_mode(table_config) == ScanPlanningMode.SERVER
@retry(**_RETRY_ARGS)
def _plan_table_scan(self, identifier: str | Identifier, request: PlanTableScanRequest) -> PlanningResponse:
"""Submit a scan plan request to the REST server.
Args:
identifier: Table identifier.
request: The scan plan request parameters.
Returns:
PlanningResponse the result of the scan plan request representing the status
Raises:
NoSuchTableError: If a table with the given identifier does not exist.
"""
self._check_endpoint(Capability.V1_SUBMIT_TABLE_SCAN_PLAN)
response = self._session.post(
self.url(Endpoints.plan_table_scan, prefixed=True, **self._split_identifier_for_path(identifier)),
data=request.model_dump_json(by_alias=True, exclude_none=True).encode(UTF8),
)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchTableError})
return _PLANNING_RESPONSE_ADAPTER.validate_json(response.text)
@retry(**_RETRY_ARGS)
def _fetch_scan_tasks(self, identifier: str | Identifier, plan_task: str) -> ScanTasks:
"""Fetch additional scan tasks using a plan task token.
Args:
identifier: Table identifier.
plan_task: The plan task token from a previous response.
Returns:
ScanTasks containing file scan tasks and possibly more plan-task tokens.
Raises:
NoSuchPlanTaskError: If a plan task with the given identifier or task does not exist.
"""
self._check_endpoint(Capability.V1_TABLE_SCAN_PLAN_TASKS)
request = FetchScanTasksRequest(plan_task=plan_task)
response = self._session.post(
self.url(Endpoints.fetch_scan_tasks, prefixed=True, **self._split_identifier_for_path(identifier)),
data=request.model_dump_json(by_alias=True).encode(UTF8),
)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchPlanTaskError})
return ScanTasks.model_validate_json(response.text)
@retry(**_RETRY_ARGS)
def _fetch_planning_result(self, identifier: str | Identifier, plan_id: str) -> PlanningResponse:
"""Fetch the result of an async scan plan by plan-id.
Args:
identifier: Table identifier.
plan_id: Plan id returned from a submitted planTableScan response.
Returns:
PlanningResponse with the current plan status.
Raises:
NoSuchPlanIdError: If the plan-id does not exist.
NoSuchTableError: If the table does not exist.
"""
self._check_endpoint(Capability.V1_FETCH_TABLE_SCAN_PLAN)
response = self._session.get(
self.url(
Endpoints.fetch_planning_result,
prefixed=True,
plan_id=quote(plan_id, safe=""),
**self._split_identifier_for_path(identifier),
),
)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchPlanIdError})
return _PLANNING_RESPONSE_ADAPTER.validate_json(response.text)
def _cancel_planning(self, identifier: str | Identifier, plan_id: str) -> bool:
"""Best-effort cancel of an async scan plan.
Returns:
True if the cancel request was accepted, False otherwise.
"""
if Capability.V1_CANCEL_TABLE_SCAN_PLAN not in self._supported_endpoints:
return False
try:
response = self._session.delete(
self.url(
Endpoints.cancel_planning,
prefixed=True,
plan_id=quote(plan_id, safe=""),
**self._split_identifier_for_path(identifier),
),
)
response.raise_for_status()
return True
except Exception:
# Plan may have already completed, failed, or been cancelled.
return False
def _poll_until_completed(self, identifier: str | Identifier, plan_id: str) -> PlanCompleted:
"""Poll fetchPlanningResult until the plan completes or times out.
Uses exponential backoff matching Java RESTTableScan defaults.
"""
max_wait_ms = property_as_int(
self.properties,
REST_SCAN_PLANNING_POLL_TIMEOUT_MS,
REST_SCAN_PLANNING_POLL_TIMEOUT_MS_DEFAULT,
)
if max_wait_ms is None or max_wait_ms <= 0:
raise ValueError(f"Invalid value for {REST_SCAN_PLANNING_POLL_TIMEOUT_MS}: {max_wait_ms} (must be positive)")
sleep_ms = float(REST_SCAN_PLANNING_POLL_MIN_SLEEP_MS)
start = time.monotonic()
retries = 0
while True:
response = self._fetch_planning_result(identifier, plan_id)
if isinstance(response, PlanCompleted):
return response
if isinstance(response, PlanFailed):
error_msg = response.error.message if response.error else "unknown error"
self._cancel_planning(identifier, plan_id)
raise RuntimeError(f"Remote scan planning failed for planId: {plan_id}: {error_msg}")
if isinstance(response, PlanCancelled):
raise RuntimeError(f"Remote scan planning cancelled for planId: {plan_id}")
if not isinstance(response, PlanSubmitted):
self._cancel_planning(identifier, plan_id)
raise RuntimeError(f"Invalid planStatus for planId: {plan_id}: {type(response).__name__}")
elapsed_ms = (time.monotonic() - start) * 1000
if retries >= REST_SCAN_PLANNING_POLL_MAX_RETRIES or elapsed_ms >= max_wait_ms:
self._cancel_planning(identifier, plan_id)
raise RemotePlanTimeoutError(
f"Remote scan planning for planId: {plan_id} did not complete within configured limits "
f"(timeout={max_wait_ms} ms, maxRetries={REST_SCAN_PLANNING_POLL_MAX_RETRIES})"
)
time.sleep(sleep_ms / 1000.0)
sleep_ms = min(sleep_ms * REST_SCAN_PLANNING_POLL_SCALE_FACTOR, REST_SCAN_PLANNING_POLL_MAX_SLEEP_MS)
retries += 1
def _expand_plan_tasks(self, identifier: str | Identifier, response: PlanCompleted) -> list[FileScanTask]:
"""Expand a completed plan response into FileScanTask objects, including pagination."""
tasks: list[FileScanTask] = []
# Collect tasks from initial response
for task in response.file_scan_tasks:
tasks.append(FileScanTask.from_rest_response(task, response.delete_files))
# Fetch and collect from additional batches
pending_tasks = deque(response.plan_tasks)
while pending_tasks:
plan_task = pending_tasks.popleft()
batch = self._fetch_scan_tasks(identifier, plan_task)
for task in batch.file_scan_tasks:
tasks.append(FileScanTask.from_rest_response(task, batch.delete_files))
pending_tasks.extend(batch.plan_tasks)
return tasks
def _plan_scan_result(self, identifier: str | Identifier, request: PlanTableScanRequest) -> PlannedScanResult:
"""Plan a table scan and return tasks with optional plan storage credentials.
Handles the full scan planning lifecycle including async polling and pagination.
"""
response = self._plan_table_scan(identifier, request)
if isinstance(response, PlanFailed):
error_msg = response.error.message if response.error else "unknown error"
raise RuntimeError(f"Received status: failed: {error_msg}")
if isinstance(response, PlanCancelled):
raise RuntimeError("Received status: cancelled")
if isinstance(response, PlanSubmitted):
if not response.plan_id:
raise ValueError("Async scan planning submitted without plan-id")
response = self._poll_until_completed(identifier, response.plan_id)
if not isinstance(response, PlanCompleted):
raise RuntimeError(f"Invalid planStatus for response: {type(response).__name__}")
tasks = self._expand_plan_tasks(identifier, response)
return PlannedScanResult(
tasks=tasks,
storage_credentials=list(response.storage_credentials or []),
plan_id=response.plan_id,
)
def plan_scan(self, identifier: str | Identifier, request: PlanTableScanRequest) -> list[FileScanTask]:
"""Plan a table scan and return FileScanTasks.
Handles the full scan planning lifecycle including async polling and pagination.
Args:
identifier: Table identifier.
request: The scan plan request parameters.
Returns:
List of FileScanTask objects ready for execution.
Raises:
RuntimeError: If planning fails, is cancelled, or returns unexpected response.
RemotePlanTimeoutError: If async planning does not complete in time.
ValueError: If a submitted plan is missing plan-id.
"""
return self._plan_scan_result(identifier, request).tasks
def _file_io_from_plan(
self,
existing_properties: Properties,
storage_credentials: list[StorageCredential],
location: str | None = None,
) -> FileIO | None:
"""Build a scan-scoped FileIO from plan storage credentials.
Layers resolved plan credentials on top of the existing scan FileIO properties so
load-time settings (for example custom S3 endpoints) are retained.
"""
if not storage_credentials:
return None
resolve_location = location
if resolve_location is None and storage_credentials[0].prefix:
resolve_location = storage_credentials[0].prefix
credential_config = self._resolve_storage_credentials(storage_credentials, resolve_location)
if not credential_config and resolve_location is None:
credential_config = dict(storage_credentials[0].config)
if not credential_config:
return None
return self._load_file_io({**existing_properties, **credential_config}, resolve_location)
def _create_legacy_oauth2_auth_manager(self, session: Session) -> AuthManager:
"""Create the LegacyOAuth2AuthManager by fetching required properties.
This will be removed in PyIceberg 1.0
"""
client_credentials = self.properties.get(CREDENTIAL)
# We want to call `self.auth_url` only when we are using CREDENTIAL
# with the legacy OAUTH2 flow as it will raise a DeprecationWarning
auth_url = self.auth_url if client_credentials is not None else None
auth_config = {
"session": session,
"auth_url": auth_url,
"credential": client_credentials,
"initial_token": self.properties.get(TOKEN),
"optional_oauth_params": self._extract_optional_oauth_params(),
}
return AuthManagerFactory.create("legacyoauth2", auth_config)
def _check_valid_namespace_identifier(self, identifier: str | Identifier) -> Identifier:
"""Check if the identifier has at least one element."""
identifier_tuple = Catalog.identifier_to_tuple(identifier)
if len(identifier_tuple) < 1:
raise NoSuchNamespaceError(f"Empty namespace identifier: {identifier}")
return identifier_tuple
def url(self, endpoint: str, prefixed: bool = True, **kwargs: Any) -> str:
"""Construct the endpoint.
Args:
endpoint: Resource identifier that points to the REST catalog.
prefixed: If the prefix return by the config needs to be appended.
Returns:
The base url of the rest catalog.
"""
url = self.uri
url = url + "v1/" if url.endswith("/") else url + "/v1/"
if prefixed:
url += self.properties.get(PREFIX, "")
url = url if url.endswith("/") else url + "/"
return url + endpoint.format(**kwargs)
def _check_endpoint(self, endpoint: Endpoint) -> None:
"""Check if an endpoint is supported by the server.
Args:
endpoint: The endpoint to check against the set of supported endpoints
Raises:
NotImplementedError: If the endpoint is not supported.
"""
if endpoint not in self._supported_endpoints:
raise NotImplementedError(f"Server does not support endpoint: {endpoint}")
@property
def auth_url(self) -> str:
self._warn_oauth_tokens_deprecation()
if url := self.properties.get(OAUTH2_SERVER_URI):
return url
else:
return self.url(Endpoints.get_token, prefixed=False)
def _warn_oauth_tokens_deprecation(self) -> None:
has_oauth_server_uri = OAUTH2_SERVER_URI in self.properties
has_credential = CREDENTIAL in self.properties
has_init_token = TOKEN in self.properties
has_sigv4_enabled = property_as_bool(self.properties, SIGV4, False)
if not has_oauth_server_uri and (has_init_token or has_credential) and not has_sigv4_enabled:
deprecation_message(
deprecated_in="0.8.0",
removed_in="1.0.0",
help_message="Iceberg REST client is missing the OAuth2 server URI "
f"configuration and defaults to {self.uri}{Endpoints.get_token}. "
"This automatic fallback will be removed in a future Iceberg release."
f"It is recommended to configure the OAuth2 endpoint using the '{OAUTH2_SERVER_URI}'"
"property to be prepared. This warning will disappear if the OAuth2"
"endpoint is explicitly configured. See https://github.com/apache/iceberg/issues/10537",
)
def _extract_optional_oauth_params(self) -> dict[str, str]:
optional_oauth_param = {SCOPE: self.properties.get(SCOPE) or CATALOG_SCOPE}
set_of_optional_params = {AUDIENCE, RESOURCE}
for param in set_of_optional_params:
if param_value := self.properties.get(param):
optional_oauth_param[param] = param_value
return optional_oauth_param
def _encode_namespace_path(self, namespace: Identifier) -> str:
"""
Encode a namespace for use as a path parameter in a URL.
Each part of the namespace is URL-encoded using `urllib.parse.quote`
(ensuring characters like '/' are encoded) and then joined by the
configured namespace separator.
"""
return self._namespace_separator.join(quote(part, safe="") for part in namespace)
def _fetch_config(self) -> None:
params = {}
if warehouse_location := self.properties.get(WAREHOUSE_LOCATION):
params[WAREHOUSE_LOCATION] = warehouse_location
with self._create_session() as session:
response = session.get(self.url(Endpoints.get_config, prefixed=False), params=params)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {})
config_response = ConfigResponse.model_validate_json(response.text)
config = config_response.defaults
config.update(self.properties)
config.update(config_response.overrides)
self.properties = config
# Update URI based on overrides
self.uri = config[URI]
# Determine supported endpoints
endpoints = config_response.endpoints
if endpoints:
self._supported_endpoints = set(endpoints)
else:
# Use default endpoints for legacy servers that don't return endpoints
self._supported_endpoints = set(DEFAULT_ENDPOINTS)
# Conditionally add view endpoints based on config
if property_as_bool(self.properties, VIEW_ENDPOINTS_SUPPORTED, VIEW_ENDPOINTS_SUPPORTED_DEFAULT):
self._supported_endpoints.update(VIEW_ENDPOINTS)
separator_from_properties = self.properties.get(NAMESPACE_SEPARATOR_PROPERTY, DEFAULT_NAMESPACE_SEPARATOR)
if not separator_from_properties:
raise ValueError("Namespace separator cannot be an empty string")
self._namespace_separator = unquote(separator_from_properties)
def _identifier_to_validated_tuple(self, identifier: str | Identifier) -> Identifier:
identifier_tuple = self.identifier_to_tuple(identifier)
if len(identifier_tuple) <= 1:
raise NoSuchIdentifierError(f"Missing namespace or invalid identifier: {'.'.join(identifier_tuple)}")
return identifier_tuple
def _split_identifier_for_path(
self, identifier: str | Identifier | TableIdentifier, kind: IdentifierKind = IdentifierKind.TABLE
) -> Properties:
if isinstance(identifier, TableIdentifier):
return {
"namespace": self._encode_namespace_path(tuple(identifier.namespace.root)),
kind.value: quote(identifier.name, safe=""),
}
identifier_tuple = self._identifier_to_validated_tuple(identifier)
# Use quote to ensure that '/' aren't treated as path separators.
return {
"namespace": self._encode_namespace_path(identifier_tuple[:-1]),
kind.value: quote(identifier_tuple[-1], safe=""),
}
def _split_identifier_for_json(self, identifier: str | Identifier) -> dict[str, Identifier | str]:
identifier_tuple = self._identifier_to_validated_tuple(identifier)
return {"namespace": identifier_tuple[:-1], "name": identifier_tuple[-1]}
def _init_sigv4(self, session: Session) -> None:
from urllib import parse
import boto3
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
class SigV4Adapter(HTTPAdapter):
def __init__(self, **properties: str):
self._properties = properties
max_retries = property_as_int(self._properties, SIGV4_MAX_RETRIES, SIGV4_MAX_RETRIES_DEFAULT)
super().__init__(max_retries=max_retries)
self._boto_session = boto3.Session(
profile_name=get_first_property_value(self._properties, AWS_PROFILE_NAME),
region_name=get_first_property_value(self._properties, AWS_REGION),
botocore_session=self._properties.get(BOTOCORE_SESSION),
aws_access_key_id=get_first_property_value(self._properties, AWS_ACCESS_KEY_ID),
aws_secret_access_key=get_first_property_value(self._properties, AWS_SECRET_ACCESS_KEY),
aws_session_token=get_first_property_value(self._properties, AWS_SESSION_TOKEN),
)
def add_headers(self, request: PreparedRequest, **kwargs: Any) -> None: # pylint: disable=W0613
credentials = self._boto_session.get_credentials().get_frozen_credentials()
region = self._properties.get(SIGV4_REGION, self._boto_session.region_name)
service = self._properties.get(SIGV4_SERVICE, "execute-api")
url = str(request.url).split("?")[0]
query = str(parse.urlsplit(request.url).query)
params = dict(parse.parse_qsl(query))
# remove the connection header as it will be updated after signing
if "connection" in request.headers:
del request.headers["connection"]
# For empty bodies, explicitly set the content hash header to the SHA256 of an empty string
if not request.body:
request.headers["x-amz-content-sha256"] = EMPTY_BODY_SHA256
aws_request = AWSRequest(
method=request.method, url=url, params=params, data=request.body, headers=dict(request.headers)
)
SigV4Auth(credentials, service, region).add_auth(aws_request)
original_header = request.headers
signed_headers = aws_request.headers
relocated_headers = {}
# relocate headers if there is a conflict with signed headers
for header, value in original_header.items():
if header in signed_headers and signed_headers[header] != value:
relocated_headers[f"Original-{header}"] = value
request.headers.update(relocated_headers)
request.headers.update(signed_headers)
session.mount(self.uri, SigV4Adapter(**self.properties))
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
)
return Table(
identifier=identifier_tuple,
metadata_location=table_response.metadata_location, # type: ignore
metadata=table_response.metadata,
io=self._load_file_io(
{**table_response.metadata.properties, **table_response.config, **credential_config},
table_response.metadata_location,
),
catalog=self,
config=table_response.config,
)
def _response_to_staged_table(self, identifier_tuple: tuple[str, ...], table_response: TableResponse) -> StagedTable:
# Per Iceberg spec: storage-credentials take precedence over config
credential_config = self._resolve_storage_credentials(
table_response.storage_credentials, table_response.metadata_location
)
return StagedTable(
identifier=identifier_tuple,
metadata_location=table_response.metadata_location, # type: ignore
metadata=table_response.metadata,
io=self._load_file_io(
{**table_response.metadata.properties, **table_response.config, **credential_config},
table_response.metadata_location,
),
catalog=self,
)
def _response_to_view(self, identifier_tuple: tuple[str, ...], view_response: ViewResponse) -> View:
return View(
identifier=identifier_tuple,
metadata=view_response.metadata,
)
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
# for backward compatibility
auth_manager = self._session.auth.auth_manager # type: ignore[union-attr]
if isinstance(auth_manager, LegacyOAuth2AuthManager):
auth_manager._refresh_token()
def _config_headers(self, session: Session) -> None:
header_properties = get_header_properties(self.properties)
session.headers.update(header_properties)
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)
def _create_table(
self,
identifier: str | Identifier,
schema: Schema | pa.Schema,
location: str | None = None,
partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
sort_order: SortOrder = UNSORTED_SORT_ORDER,
properties: Properties = EMPTY_DICT,
stage_create: bool = False,
) -> TableResponse:
self._check_endpoint(Capability.V1_CREATE_TABLE)
iceberg_schema = self._convert_schema_if_needed(
schema,
int(properties.get(TableProperties.FORMAT_VERSION, TableProperties.DEFAULT_FORMAT_VERSION)), # type: ignore
)
fresh_schema = assign_fresh_schema_ids(iceberg_schema)
fresh_partition_spec = assign_fresh_partition_spec_ids(partition_spec, iceberg_schema, fresh_schema)
fresh_sort_order = assign_fresh_sort_order_ids(sort_order, iceberg_schema, fresh_schema)
namespace_and_table = self._split_identifier_for_path(identifier)
if location:
location = location.rstrip("/")
request = CreateTableRequest(
name=self._identifier_to_validated_tuple(identifier)[-1],
location=location,
table_schema=fresh_schema,
partition_spec=fresh_partition_spec,
write_order=fresh_sort_order,
stage_create=stage_create,
properties=properties,
)
serialized_json = request.model_dump_json().encode(UTF8)
response = self._session.post(
self.url(Endpoints.create_table, namespace=namespace_and_table["namespace"]),
data=serialized_json,
)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {409: TableAlreadyExistsError, 404: NoSuchNamespaceError})
return TableResponse.model_validate_json(response.text)
@override
@retry(**_RETRY_ARGS)
def create_table(
self,
identifier: str | Identifier,
schema: Schema | pa.Schema,
location: str | None = None,
partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
sort_order: SortOrder = UNSORTED_SORT_ORDER,
properties: Properties = EMPTY_DICT,
) -> Table:
table_response = self._create_table(
identifier=identifier,
schema=schema,
location=location,
partition_spec=partition_spec,
sort_order=sort_order,
properties=properties,
stage_create=False,
)
return self._response_to_table(self.identifier_to_tuple(identifier), table_response)
@override
@retry(**_RETRY_ARGS)
def create_table_transaction(
self,
identifier: str | Identifier,
schema: Schema | pa.Schema,
location: str | None = None,
partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
sort_order: SortOrder = UNSORTED_SORT_ORDER,
properties: Properties = EMPTY_DICT,
) -> CreateTableTransaction:
table_response = self._create_table(
identifier=identifier,
schema=schema,
location=location,
partition_spec=partition_spec,
sort_order=sort_order,
properties=properties,
stage_create=True,
)
staged_table = self._response_to_staged_table(self.identifier_to_tuple(identifier), table_response)
return CreateTableTransaction(staged_table)
@override
@retry(**_RETRY_ARGS)
def create_view(
self,
identifier: str | Identifier,
schema: Schema | pa.Schema,
view_version: ViewVersion,
location: str | None = None,
properties: Properties = EMPTY_DICT,
) -> View:
iceberg_schema = self._convert_schema_if_needed(schema)
fresh_schema = assign_fresh_schema_ids(iceberg_schema)
namespace_and_view = self._split_identifier_for_path(identifier, IdentifierKind.VIEW)
if location:
location = location.rstrip("/")
request = CreateViewRequest(
name=namespace_and_view["view"],
location=location,
view_schema=fresh_schema,
view_version=view_version,
properties=properties,
)
serialized_json = request.model_dump_json().encode(UTF8)
response = self._session.post(
self.url(Endpoints.create_view, namespace=namespace_and_view["namespace"]),
data=serialized_json,
)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {409: ViewAlreadyExistsError})
view_response = ViewResponse.model_validate_json(response.text)
return self._response_to_view(self.identifier_to_tuple(identifier), view_response)
@retry(**_RETRY_ARGS)
@override
def register_table(self, identifier: str | Identifier, metadata_location: str, overwrite: bool = False) -> Table:
"""Register a new table using existing metadata.
Args:
identifier (Union[str, Identifier]): Table identifier for the table
metadata_location (str): The location to the metadata
overwrite (bool): Whether to overwrite the existing table, default False
Returns:
Table: The newly registered table
Raises:
TableAlreadyExistsError: If the table already exists
"""
self._check_endpoint(Capability.V1_REGISTER_TABLE)
namespace_and_table = self._split_identifier_for_path(identifier)
request = RegisterTableRequest(
name=self._identifier_to_validated_tuple(identifier)[-1],
metadata_location=metadata_location,
overwrite=overwrite,
)
serialized_json = request.model_dump_json().encode(UTF8)
response = self._session.post(
self.url(Endpoints.register_table, namespace=namespace_and_table["namespace"]),
data=serialized_json,
)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {409: TableAlreadyExistsError})
table_response = TableResponse.model_validate_json(response.text)
return self._response_to_table(self.identifier_to_tuple(identifier), table_response)
@retry(**_RETRY_ARGS)
@override
def list_tables(self, namespace: str | Identifier) -> list[Identifier]:
self._check_endpoint(Capability.V1_LIST_TABLES)
namespace_tuple = self._check_valid_namespace_identifier(namespace)
namespace_concat = self._encode_namespace_path(namespace_tuple)
url = self.url(Endpoints.list_tables, namespace=namespace_concat)
params: dict[str, str] = {}
page_size = property_as_int(self.properties, PAGE_SIZE, None)
if page_size is not None:
if page_size <= 0:
raise ValueError(f"{PAGE_SIZE} must be a positive integer")
params["pageSize"] = str(page_size)
tables: list[Identifier] = []
page_token: str | None = None
while True:
if page_token:
params["pageToken"] = page_token
response = self._session.get(url, params=params)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchNamespaceError})
parsed = ListTablesResponse.model_validate_json(response.text)
tables.extend([(*table.namespace, table.name) for table in parsed.identifiers])
if not parsed.next_page_token:
break
page_token = parsed.next_page_token
return tables
@retry(**_RETRY_ARGS)
@override
def load_table(self, identifier: str | Identifier) -> Table:
self._check_endpoint(Capability.V1_LOAD_TABLE)
params = {}
if mode := self.properties.get(SNAPSHOT_LOADING_MODE):
if mode in {"all", "refs"}:
params["snapshots"] = mode
else:
raise ValueError("Invalid snapshot-loading-mode: {}")
response = self._session.get(
self.url(Endpoints.load_table, prefixed=True, **self._split_identifier_for_path(identifier)), params=params
)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchTableError})
table_response = TableResponse.model_validate_json(response.text)
return self._response_to_table(self.identifier_to_tuple(identifier), table_response)
@retry(**_RETRY_ARGS)
def _load_credentials(
self,
identifier: str | Identifier,
) -> LoadCredentialsResponse:
"""Load raw vended storage credentials for a table."""
self._check_endpoint(Capability.V1_LOAD_CREDENTIALS)
response = self._session.get(
self.url(Endpoints.load_credentials, prefixed=True, **self._split_identifier_for_path(identifier)),
)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchTableError})
return LoadCredentialsResponse.model_validate_json(response.text)
def load_credentials(
self,
identifier: str | Identifier,
location: str,
) -> Properties:
"""Load vended storage credentials and return the best match for a location."""
credentials_response = self._load_credentials(identifier)
return self._resolve_storage_credentials(credentials_response.storage_credentials, location)
@retry(**_RETRY_ARGS)
@override
def drop_table(self, identifier: str | Identifier, purge_requested: bool = False) -> None:
self._check_endpoint(Capability.V1_DELETE_TABLE)
response = self._session.delete(
self.url(Endpoints.drop_table, prefixed=True, **self._split_identifier_for_path(identifier)),
params={"purgeRequested": purge_requested},
)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchTableError})
@retry(**_RETRY_ARGS)
@override
def purge_table(self, identifier: str | Identifier) -> None:
self.drop_table(identifier=identifier, purge_requested=True)
@retry(**_RETRY_ARGS)
@override
def rename_table(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> Table:
self._check_endpoint(Capability.V1_RENAME_TABLE)
payload = {
"source": self._split_identifier_for_json(from_identifier),
"destination": self._split_identifier_for_json(to_identifier),
}
# Ensure that namespaces exist on source and destination.
source_namespace = self._split_identifier_for_json(from_identifier)["namespace"]
if not self.namespace_exists(source_namespace):
raise NoSuchNamespaceError(f"Source namespace does not exist: {source_namespace}")
destination_namespace = self._split_identifier_for_json(to_identifier)["namespace"]
if not self.namespace_exists(destination_namespace):
raise NoSuchNamespaceError(f"Destination namespace does not exist: {destination_namespace}")
response = self._session.post(self.url(Endpoints.rename_table), json=payload)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchTableError, 409: TableAlreadyExistsError})
return self.load_table(to_identifier)
def _remove_catalog_name_from_table_request_identifier(self, table_request: CommitTableRequest) -> CommitTableRequest:
if table_request.identifier.namespace.root[0] == self.name:
return table_request.model_copy(
update={
"identifier": TableIdentifier(
namespace=table_request.identifier.namespace.root[1:], name=table_request.identifier.name
)
}
)
return table_request
@retry(**_RETRY_ARGS)
@override
def list_views(self, namespace: str | Identifier) -> list[Identifier]:
if Capability.V1_LIST_VIEWS not in self._supported_endpoints:
return []
namespace_tuple = self._check_valid_namespace_identifier(namespace)
namespace_concat = self._encode_namespace_path(namespace_tuple)
url = self.url(Endpoints.list_views, namespace=namespace_concat)
params: dict[str, str] = {}
page_size = property_as_int(self.properties, PAGE_SIZE, None)
if page_size is not None:
if page_size <= 0:
raise ValueError(f"{PAGE_SIZE} must be a positive integer")
params["pageSize"] = str(page_size)
views: list[Identifier] = []
page_token: str | None = None
while True:
if page_token:
params["pageToken"] = page_token
response = self._session.get(url, params=params)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchNamespaceError})
parsed = ListViewsResponse.model_validate_json(response.text)
views.extend([(*view.namespace, view.name) for view in parsed.identifiers])
if not parsed.next_page_token:
break
page_token = parsed.next_page_token
return views
@retry(**_RETRY_ARGS)
@override
def load_view(self, identifier: str | Identifier) -> View:
self._check_endpoint(Capability.V1_LOAD_VIEW)
response = self._session.get(
self.url(Endpoints.load_view, prefixed=True, **self._split_identifier_for_path(identifier, IdentifierKind.VIEW))
)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchViewError})
view_response = ViewResponse.model_validate_json(response.text)
return self._response_to_view(self.identifier_to_tuple(identifier), view_response)
@retry(**_RETRY_ARGS)
@override
def commit_table(
self, table: Table, requirements: tuple[TableRequirement, ...], updates: tuple[TableUpdate, ...]
) -> CommitTableResponse:
"""Commit updates to a table.
Args:
table (Table): The table to be updated.
requirements: (Tuple[TableRequirement, ...]): Table requirements.
updates: (Tuple[TableUpdate, ...]): Table updates.
Returns:
CommitTableResponse: The updated metadata.
Raises:
NoSuchTableError: If a table with the given identifier does not exist.
CommitFailedException: Requirement not met, or a conflict with a concurrent commit.
CommitStateUnknownException: Failed due to an internal exception on the side of the catalog.
"""
self._check_endpoint(Capability.V1_UPDATE_TABLE)
identifier = table.name()
table_identifier = TableIdentifier(namespace=identifier[:-1], name=identifier[-1])
table_request = CommitTableRequest(identifier=table_identifier, requirements=requirements, updates=updates)
headers = self._session.headers
if table_token := table.config.get(TOKEN):
headers[AUTHORIZATION_HEADER] = f"{BEARER_PREFIX} {table_token}"
response = self._session.post(
self.url(Endpoints.update_table, prefixed=True, **self._split_identifier_for_path(table_request.identifier)),
data=table_request.model_dump_json().encode(UTF8),
headers=headers,
)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(
exc,
{
409: CommitFailedException,
500: CommitStateUnknownException,
502: CommitStateUnknownException,
504: CommitStateUnknownException,
},
)
return CommitTableResponse.model_validate_json(response.text)
@retry(**_RETRY_ARGS)
@override
def create_namespace(self, namespace: str | Identifier, properties: Properties = EMPTY_DICT) -> None:
self._check_endpoint(Capability.V1_CREATE_NAMESPACE)
namespace_tuple = self._check_valid_namespace_identifier(namespace)
payload = {"namespace": namespace_tuple, "properties": properties}
response = self._session.post(self.url(Endpoints.create_namespace), json=payload)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {409: NamespaceAlreadyExistsError})
@retry(**_RETRY_ARGS)
@override
def drop_namespace(self, namespace: str | Identifier) -> None:
self._check_endpoint(Capability.V1_DELETE_NAMESPACE)
namespace_tuple = self._check_valid_namespace_identifier(namespace)
namespace = self._encode_namespace_path(namespace_tuple)
response = self._session.delete(self.url(Endpoints.drop_namespace, namespace=namespace))
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchNamespaceError, 409: NamespaceNotEmptyError})
@retry(**_RETRY_ARGS)
@override
def list_namespaces(self, namespace: str | Identifier = ()) -> list[Identifier]:
self._check_endpoint(Capability.V1_LIST_NAMESPACES)
namespace_tuple = self.identifier_to_tuple(namespace)
params: dict[str, str] = {}
page_size = property_as_int(self.properties, PAGE_SIZE, None)
if page_size is not None:
if page_size <= 0:
raise ValueError(f"{PAGE_SIZE} must be a positive integer")
params["pageSize"] = str(page_size)
namespaces: list[Identifier] = []
page_token: str | None = None
while True:
if namespace_tuple:
params["parent"] = self._encode_namespace_path(namespace_tuple)
if page_token:
params["pageToken"] = page_token
response = self._session.get(self.url(Endpoints.list_namespaces), params=params)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchNamespaceError})
parsed = ListNamespaceResponse.model_validate_json(response.text)
namespaces.extend(parsed.namespaces)
if not parsed.next_page_token:
break
page_token = parsed.next_page_token
return namespaces
@retry(**_RETRY_ARGS)
@override
def load_namespace_properties(self, namespace: str | Identifier) -> Properties:
self._check_endpoint(Capability.V1_LOAD_NAMESPACE)
namespace_tuple = self._check_valid_namespace_identifier(namespace)
namespace = self._encode_namespace_path(namespace_tuple)
response = self._session.get(self.url(Endpoints.load_namespace_metadata, namespace=namespace))
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchNamespaceError})
return NamespaceResponse.model_validate_json(response.text).properties
@retry(**_RETRY_ARGS)
@override
def update_namespace_properties(
self, namespace: str | Identifier, removals: set[str] | None = None, updates: Properties = EMPTY_DICT
) -> PropertiesUpdateSummary:
self._check_endpoint(Capability.V1_UPDATE_NAMESPACE)
namespace_tuple = self._check_valid_namespace_identifier(namespace)
namespace = self._encode_namespace_path(namespace_tuple)
payload = {"removals": list(removals or []), "updates": updates}
response = self._session.post(self.url(Endpoints.update_namespace_properties, namespace=namespace), json=payload)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchNamespaceError})
parsed_response = UpdateNamespacePropertiesResponse.model_validate_json(response.text)
return PropertiesUpdateSummary(
removed=parsed_response.removed,
updated=parsed_response.updated,
missing=parsed_response.missing,
)
@retry(**_RETRY_ARGS)
@override
def namespace_exists(self, namespace: str | Identifier) -> bool:
namespace_tuple = self._check_valid_namespace_identifier(namespace)
namespace = self._encode_namespace_path(namespace_tuple)
# fallback in order to work with older rest catalog implementations
if Capability.V1_NAMESPACE_EXISTS not in self._supported_endpoints:
try:
self.load_namespace_properties(namespace_tuple)
return True
except NoSuchNamespaceError:
return False
response = self._session.head(self.url(Endpoints.namespace_exists, namespace=namespace))
if response.status_code == 404:
return False
elif response.status_code in (200, 204):
return True
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {})
return False
@retry(**_RETRY_ARGS)
@override
def table_exists(self, identifier: str | Identifier) -> bool:
"""Check if a table exists.
Args:
identifier (str | Identifier): Table identifier.
Returns:
bool: True if the table exists, False otherwise.
"""
# fallback in order to work with older rest catalog implementations
if Capability.V1_TABLE_EXISTS not in self._supported_endpoints:
try:
self.load_table(identifier)
return True
except NoSuchTableError:
return False
response = self._session.head(
self.url(Endpoints.load_table, prefixed=True, **self._split_identifier_for_path(identifier))
)
if response.status_code == 404:
return False
elif response.status_code in (200, 204):
return True
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {})
return False
@retry(**_RETRY_ARGS)
@override
def view_exists(self, identifier: str | Identifier) -> bool:
"""Check if a view exists.
Args:
identifier (str | Identifier): View identifier.
Returns:
bool: True if the view exists, False otherwise.
"""
response = self._session.head(
self.url(Endpoints.view_exists, prefixed=True, **self._split_identifier_for_path(identifier, IdentifierKind.VIEW)),
)
if response.status_code == 404:
return False
elif response.status_code in [200, 204]:
return True
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {})
return False
@retry(**_RETRY_ARGS)
@override
def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
self._check_endpoint(Capability.V1_REGISTER_VIEW)
namespace_and_view = self._split_identifier_for_path(identifier, IdentifierKind.VIEW)
namespace = namespace_and_view["namespace"]
view = namespace_and_view["view"]
if self.table_exists(identifier):
raise TableAlreadyExistsError(f"Table {namespace}.{view} already exists")
request = RegisterViewRequest(name=view, metadata_location=metadata_location)
serialized_json = request.model_dump_json().encode(UTF8)
response = self._session.post(
self.url(Endpoints.register_view, namespace=namespace),
data=serialized_json,
)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {409: ViewAlreadyExistsError})
view_response = ViewResponse.model_validate_json(response.text)
return self._response_to_view(self.identifier_to_tuple(identifier), view_response)
@retry(**_RETRY_ARGS)
@override
def drop_view(self, identifier: str | Identifier) -> None:
self._check_endpoint(Capability.V1_DELETE_VIEW)
response = self._session.delete(
self.url(Endpoints.drop_view, prefixed=True, **self._split_identifier_for_path(identifier, IdentifierKind.VIEW)),
)
try:
response.raise_for_status()
except HTTPError as exc:
_handle_non_200_response(exc, {404: NoSuchViewError})
def close(self) -> None:
"""Close the catalog and release Session connection adapters.
This method closes mounted HttpAdapters' pooled connections and any active Proxy pooled connections.
"""
self._session.close()