REST: Add support for unregister_table

pyiceberg/catalog/rest/__init__.py

@@ -149,6 +149,7 @@ class Endpoints:
     load_table: str = "namespaces/{namespace}/tables/{table}"
     update_table: str = "namespaces/{namespace}/tables/{table}"
     drop_table: str = "namespaces/{namespace}/tables/{table}"
+    unregister_table: str = "namespaces/{namespace}/tables/{table}/unregister"
     table_exists: str = "namespaces/{namespace}/tables/{table}"
     get_token: str = "oauth/tokens"
     rename_table: str = "tables/rename"
@@ -181,6 +182,7 @@ class Capability:
     V1_DELETE_TABLE = Endpoint(http_method=HttpMethod.DELETE, path=f"{API_PREFIX}/{Endpoints.drop_table}")
     V1_RENAME_TABLE = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.rename_table}")
     V1_REGISTER_TABLE = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.register_table}")
+    V1_UNREGISTER_TABLE = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.unregister_table}")
 
     V1_LIST_VIEWS = Endpoint(http_method=HttpMethod.GET, path=f"{API_PREFIX}/{Endpoints.list_views}")
     V1_LOAD_VIEW = Endpoint(http_method=HttpMethod.GET, path=f"{API_PREFIX}/{Endpoints.load_view}")
@@ -332,6 +334,16 @@ class RegisterTableRequest(IcebergBaseModel):
     overwrite: bool
 
 
+class UnregisterTableResult(IcebergBaseModel):
+    """Result of unregistering a table.
+
+    Contains the last metadata location and table metadata at the time of unregistration.
+    """
+
+    metadata_location: str = Field(..., alias="metadata-location")
+    metadata: TableMetadata
+
+
 class RegisterViewRequest(IcebergBaseModel):
     name: str
     metadata_location: str = Field(..., alias="metadata-location")
@@ -1036,6 +1048,32 @@ def register_table(self, identifier: str | Identifier, metadata_location: str, o
         table_response = TableResponse.model_validate_json(response.text)
         return self._response_to_table(self.identifier_to_tuple(identifier), table_response)
 
+    @retry(**_RETRY_ARGS)
+    def unregister_table(self, identifier: str | Identifier) -> tuple[str, TableMetadata]:
+        """Unregister a table from the catalog without removing data or metadata files.
+
+        Args:
+            identifier (Union[str, Identifier]): Table identifier for the table
+
+        Returns:
+            tuple[str, TableMetadata]: The last metadata location and corresponding table metadata
+
+        Raises:
+            NoSuchTableError: If the table does not exist
+        """
+        self._check_endpoint(Capability.V1_UNREGISTER_TABLE)
+        namespace_and_table = self._split_identifier_for_path(identifier)
+        response = self._session.post(
+            self.url(Endpoints.unregister_table, prefixed=True, **namespace_and_table),
+        )
+        try:
+            response.raise_for_status()
+        except HTTPError as exc:
+            _handle_non_200_response(exc, {404: NoSuchTableError})
+
+        result = UnregisterTableResult.model_validate_json(response.content)
+        return (result.metadata_location, result.metadata)
+
     @retry(**_RETRY_ARGS)
     @override
     def list_tables(self, namespace: str | Identifier) -> list[Identifier]:

tests/catalog/test_rest.py

@@ -104,6 +104,7 @@
     Capability.V1_DELETE_TABLE,
     Capability.V1_RENAME_TABLE,
     Capability.V1_REGISTER_TABLE,
+    Capability.V1_UNREGISTER_TABLE,
     Capability.V1_LIST_VIEWS,
     Capability.V1_LOAD_VIEW,
     Capability.V1_VIEW_EXISTS,
@@ -1997,6 +1998,46 @@ def test_register_table_overwrite(
     assert actual.name() == expected.name()
 
 
+def test_unregister_table_200(
+    rest_mock: Mocker, table_schema_simple: Schema, example_table_metadata_no_snapshot_v1_rest_json: dict[str, Any]
+) -> None:
+    unregister_response = {
+        "metadata-location": "s3://warehouse/database/table/metadata.json",
+        "metadata": example_table_metadata_no_snapshot_v1_rest_json["metadata"],
+    }
+    rest_mock.post(
+        f"{TEST_URI}v1/namespaces/default/tables/my_table/unregister",
+        json=unregister_response,
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    metadata_location, metadata = catalog.unregister_table(identifier=("default", "my_table"))
+
+    assert metadata_location == "s3://warehouse/database/table/metadata.json"
+    assert metadata.model_dump() == TableMetadataV1(**example_table_metadata_no_snapshot_v1_rest_json["metadata"]).model_dump()
+
+
+def test_unregister_table_404(rest_mock: Mocker) -> None:
+    rest_mock.post(
+        f"{TEST_URI}v1/namespaces/default/tables/does_not_exist/unregister",
+        json={
+            "error": {
+                "message": "Table does not exist: default.does_not_exist",
+                "type": "NoSuchTableException",
+                "code": 404,
+            }
+        },
+        status_code=404,
+        request_headers=TEST_HEADERS,
+    )
+
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    with pytest.raises(NoSuchTableError) as e:
+        catalog.unregister_table(identifier=("default", "does_not_exist"))
+    assert "Table does not exist" in str(e.value)
+
+
 def test_delete_namespace_204(rest_mock: Mocker) -> None:
     namespace = "example"
     rest_mock.delete(