Add drop_view to the rest catalog (#820)

* Use NoSuchIdentifier instead of NoTableIdentifier

This change allows for the validation of both Tables and Views.

* Add drop_view to the rest catalog

* fixup! Add drop_view to the rest catalog

pyiceberg/catalog/__init__.py

@@ -629,6 +629,17 @@ def update_namespace_properties(
             ValueError: If removals and updates have overlapping keys.
         """
 
+    @abstractmethod
+    def drop_view(self, identifier: Union[str, Identifier]) -> None:
+        """Drop a view.
+
+        Args:
+            identifier (str | Identifier): View identifier.
+
+        Raises:
+            NoSuchViewError: If a view with the given name does not exist.
+        """
+
     @deprecated(
         deprecated_in="0.8.0",
         removed_in="0.9.0",

pyiceberg/catalog/dynamodb.py

@@ -531,6 +531,9 @@ def update_namespace_properties(
     def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
         raise NotImplementedError
 
+    def drop_view(self, identifier: Union[str, Identifier]) -> None:
+        raise NotImplementedError
+
     def _get_iceberg_table_item(self, database_name: str, table_name: str) -> Dict[str, Any]:
         try:
             return self._get_dynamo_item(identifier=f"{database_name}.{table_name}", namespace=database_name)

pyiceberg/catalog/glue.py

@@ -772,3 +772,6 @@ def update_namespace_properties(
 
     def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
         raise NotImplementedError
+
+    def drop_view(self, identifier: Union[str, Identifier]) -> None:
+        raise NotImplementedError

pyiceberg/catalog/hive.py

@@ -707,3 +707,6 @@ def update_namespace_properties(
         expected_to_change = (removals or set()).difference(removed)
 
         return PropertiesUpdateSummary(removed=list(removed or []), updated=list(updated or []), missing=list(expected_to_change))
+
+    def drop_view(self, identifier: Union[str, Identifier]) -> None:
+        raise NotImplementedError

pyiceberg/catalog/noop.py

@@ -116,3 +116,6 @@ def update_namespace_properties(
 
     def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
         raise NotImplementedError
+
+    def drop_view(self, identifier: Union[str, Identifier]) -> None:
+        raise NotImplementedError

pyiceberg/catalog/rest.py

@@ -14,6 +14,7 @@
 #  KIND, either express or implied.  See the License for the
 #  specific language governing permissions and limitations
 #  under the License.
+from enum import Enum
 from json import JSONDecodeError
 from typing import (
     TYPE_CHECKING,
@@ -48,8 +49,10 @@
     ForbiddenError,
     NamespaceAlreadyExistsError,
     NamespaceNotEmptyError,
+    NoSuchIdentifierError,
     NoSuchNamespaceError,
     NoSuchTableError,
+    NoSuchViewError,
     OAuthError,
     RESTError,
     ServerError,
@@ -97,6 +100,12 @@ class Endpoints:
     get_token: str = "oauth/tokens"
     rename_table: str = "tables/rename"
     list_views: str = "namespaces/{namespace}/views"
+    drop_view: str = "namespaces/{namespace}/views/{view}"
+
+
+class IdentifierKind(Enum):
+    TABLE = "table"
+    VIEW = "view"
 
 
 AUTHORIZATION_HEADER = "Authorization"
@@ -389,17 +398,20 @@ def _fetch_config(self) -> None:
     def _identifier_to_validated_tuple(self, identifier: Union[str, Identifier]) -> Identifier:
         identifier_tuple = self.identifier_to_tuple(identifier)
         if len(identifier_tuple) <= 1:
-            raise NoSuchTableError(f"Missing namespace or invalid identifier: {'.'.join(identifier_tuple)}")
+            raise NoSuchIdentifierError(f"Missing namespace or invalid identifier: {'.'.join(identifier_tuple)}")
         return identifier_tuple
 
-    def _split_identifier_for_path(self, identifier: Union[str, Identifier, TableIdentifier]) -> Properties:
+    def _split_identifier_for_path(
+        self, identifier: Union[str, Identifier, TableIdentifier], kind: IdentifierKind = IdentifierKind.TABLE
+    ) -> Properties:
         if isinstance(identifier, TableIdentifier):
             if identifier.namespace.root[0] == self.name:
-                return {"namespace": NAMESPACE_SEPARATOR.join(identifier.namespace.root[1:]), "table": identifier.name}
+                return {"namespace": NAMESPACE_SEPARATOR.join(identifier.namespace.root[1:]), kind.value: identifier.name}
             else:
-                return {"namespace": NAMESPACE_SEPARATOR.join(identifier.namespace.root), "table": identifier.name}
+                return {"namespace": NAMESPACE_SEPARATOR.join(identifier.namespace.root), kind.value: identifier.name}
         identifier_tuple = self._identifier_to_validated_tuple(identifier)
-        return {"namespace": NAMESPACE_SEPARATOR.join(identifier_tuple[:-1]), "table": identifier_tuple[-1]}
+
+        return {"namespace": NAMESPACE_SEPARATOR.join(identifier_tuple[:-1]), kind.value: identifier_tuple[-1]}
 
     def _split_identifier_for_json(self, identifier: Union[str, Identifier]) -> Dict[str, Union[Identifier, str]]:
         identifier_tuple = self._identifier_to_validated_tuple(identifier)
@@ -867,3 +879,16 @@ def table_exists(self, identifier: Union[str, Identifier]) -> bool:
             self._handle_non_200_response(exc, {})
 
         return False
+
+    @retry(**_RETRY_ARGS)
+    def drop_view(self, identifier: Union[str]) -> None:
+        identifier_tuple = self.identifier_to_tuple_without_catalog(identifier)
+        response = self._session.delete(
+            self.url(
+                Endpoints.drop_view, prefixed=True, **self._split_identifier_for_path(identifier_tuple, IdentifierKind.VIEW)
+            ),
+        )
+        try:
+            response.raise_for_status()
+        except HTTPError as exc:
+            self._handle_non_200_response(exc, {404: NoSuchViewError})

pyiceberg/catalog/sql.py

@@ -699,3 +699,6 @@ def update_namespace_properties(
 
     def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
         raise NotImplementedError
+
+    def drop_view(self, identifier: Union[str, Identifier]) -> None:
+        raise NotImplementedError

pyiceberg/exceptions.py

@@ -40,6 +40,14 @@ class NoSuchIcebergTableError(NoSuchTableError):
     """Raises when the table found in the REST catalog is not an iceberg table."""
 
 
+class NoSuchViewError(Exception):
+    """Raises when the view can't be found in the REST catalog."""
+
+
+class NoSuchIdentifierError(Exception):
+    """Raises when the identifier can't be found in the REST catalog."""
+
+
 class NoSuchNamespaceError(Exception):
     """Raised when a referenced name-space is not found."""
 

tests/catalog/test_base.py

@@ -259,6 +259,9 @@ def update_namespace_properties(
     def list_views(self, namespace: Optional[Union[str, Identifier]] = None) -> List[Identifier]:
         raise NotImplementedError
 
+    def drop_view(self, identifier: Union[str, Identifier]) -> None:
+        raise NotImplementedError
+
 
 @pytest.fixture
 def catalog(tmp_path: PosixPath) -> InMemoryCatalog:

tests/catalog/test_rest.py

@@ -29,8 +29,10 @@
     AuthorizationExpiredError,
     NamespaceAlreadyExistsError,
     NamespaceNotEmptyError,
+    NoSuchIdentifierError,
     NoSuchNamespaceError,
     NoSuchTableError,
+    NoSuchViewError,
     OAuthError,
     ServerError,
     TableAlreadyExistsError,
@@ -1158,31 +1160,31 @@ def test_delete_table_404(rest_mock: Mocker) -> None:
 
 def test_create_table_missing_namespace(rest_mock: Mocker, table_schema_simple: Schema) -> None:
     table = "table"
-    with pytest.raises(NoSuchTableError) as e:
+    with pytest.raises(NoSuchIdentifierError) as e:
         # Missing namespace
         RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).create_table(table, table_schema_simple)
     assert f"Missing namespace or invalid identifier: {table}" in str(e.value)
 
 
 def test_load_table_invalid_namespace(rest_mock: Mocker) -> None:
     table = "table"
-    with pytest.raises(NoSuchTableError) as e:
+    with pytest.raises(NoSuchIdentifierError) as e:
         # Missing namespace
         RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).load_table(table)
     assert f"Missing namespace or invalid identifier: {table}" in str(e.value)
 
 
 def test_drop_table_invalid_namespace(rest_mock: Mocker) -> None:
     table = "table"
-    with pytest.raises(NoSuchTableError) as e:
+    with pytest.raises(NoSuchIdentifierError) as e:
         # Missing namespace
         RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).drop_table(table)
     assert f"Missing namespace or invalid identifier: {table}" in str(e.value)
 
 
 def test_purge_table_invalid_namespace(rest_mock: Mocker) -> None:
     table = "table"
-    with pytest.raises(NoSuchTableError) as e:
+    with pytest.raises(NoSuchIdentifierError) as e:
         # Missing namespace
         RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).purge_table(table)
     assert f"Missing namespace or invalid identifier: {table}" in str(e.value)
@@ -1307,3 +1309,41 @@ def test_table_identifier_in_commit_table_request(rest_mock: Mocker, example_tab
         rest_mock.last_request.text
         == """{"identifier":{"namespace":["namespace"],"name":"table_name"},"requirements":[],"updates":[]}"""
     )
+
+
+def test_drop_view_invalid_namespace(rest_mock: Mocker) -> None:
+    view = "view"
+    with pytest.raises(NoSuchIdentifierError) as e:
+        # Missing namespace
+        RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).drop_view(view)
+
+    assert f"Missing namespace or invalid identifier: {view}" in str(e.value)
+
+
+def test_drop_view_404(rest_mock: Mocker) -> None:
+    rest_mock.delete(
+        f"{TEST_URI}v1/namespaces/some_namespace/views/does_not_exists",
+        json={
+            "error": {
+                "message": "The given view does not exist",
+                "type": "NoSuchViewException",
+                "code": 404,
+            }
+        },
+        status_code=404,
+        request_headers=TEST_HEADERS,
+    )
+
+    with pytest.raises(NoSuchViewError) as e:
+        RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).drop_view(("some_namespace", "does_not_exists"))
+    assert "The given view does not exist" in str(e.value)
+
+
+def test_drop_view_204(rest_mock: Mocker) -> None:
+    rest_mock.delete(
+        f"{TEST_URI}v1/namespaces/some_namespace/views/some_view",
+        json={},
+        status_code=204,
+        request_headers=TEST_HEADERS,
+    )
+    RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).drop_view(("some_namespace", "some_view"))