Handle error when trying to drop a non-empty namespace on rest.py (#868)

Author: uatach

pyiceberg/catalog/rest.py

@@ -47,6 +47,7 @@
     CommitStateUnknownException,
     ForbiddenError,
     NamespaceAlreadyExistsError,
+    NamespaceNotEmptyError,
     NoSuchNamespaceError,
     NoSuchTableError,
     OAuthError,
@@ -725,7 +726,7 @@ def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
         try:
             response.raise_for_status()
         except HTTPError as exc:
-            self._handle_non_200_response(exc, {404: NoSuchNamespaceError})
+            self._handle_non_200_response(exc, {404: NoSuchNamespaceError, 409: NamespaceNotEmptyError})
 
     @retry(**_RETRY_ARGS)
     def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identifier]:

tests/catalog/test_rest.py

@@ -28,6 +28,7 @@
 from pyiceberg.exceptions import (
     AuthorizationExpiredError,
     NamespaceAlreadyExistsError,
+    NamespaceNotEmptyError,
     NoSuchNamespaceError,
     NoSuchTableError,
     OAuthError,
@@ -556,6 +557,25 @@ def test_drop_namespace_404(rest_mock: Mocker) -> None:
     assert "Namespace does not exist" in str(e.value)
 
 
+def test_drop_namespace_409(rest_mock: Mocker) -> None:
+    namespace = "examples"
+    rest_mock.delete(
+        f"{TEST_URI}v1/namespaces/{namespace}",
+        json={
+            "error": {
+                "message": "Namespace is not empty: leden in warehouse 8bcb0838-50fc-472d-9ddb-8feb89ef5f1e",
+                "type": "NamespaceNotEmptyError",
+                "code": 409,
+            }
+        },
+        status_code=409,
+        request_headers=TEST_HEADERS,
+    )
+    with pytest.raises(NamespaceNotEmptyError) as e:
+        RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).drop_namespace(namespace)
+    assert "Namespace is not empty" in str(e.value)
+
+
 def test_load_namespace_properties_200(rest_mock: Mocker) -> None:
     namespace = "leden"
     rest_mock.get(