Implementing namespace_exists function on the REST Catalog (apache#1434)

* - Added the namespace_exists function in the RESTCatalog
- Added the relevant unit tests

* - Removed docstring to match other namespace functions

* - Added integration test for REST Catalog namespace_exists functionality

* - Added ASF license to test_rest_catalog.py to recover from failing test

pyiceberg/catalog/rest.py

@@ -94,6 +94,7 @@ class Endpoints:
     load_namespace_metadata: str = "namespaces/{namespace}"
     drop_namespace: str = "namespaces/{namespace}"
     update_namespace_properties: str = "namespaces/{namespace}/properties"
+    namespace_exists: str = "namespaces/{namespace}"
     list_tables: str = "namespaces/{namespace}/tables"
     create_table: str = "namespaces/{namespace}/tables"
     register_table = "namespaces/{namespace}/register"
@@ -870,6 +871,24 @@ def update_namespace_properties(
             missing=parsed_response.missing,
         )
 
+    @retry(**_RETRY_ARGS)
+    def namespace_exists(self, namespace: Union[str, Identifier]) -> bool:
+        namespace_tuple = self._check_valid_namespace_identifier(namespace)
+        namespace = NAMESPACE_SEPARATOR.join(namespace_tuple)
+        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:
+            self._handle_non_200_response(exc, {})
+
+        return False
+
     @retry(**_RETRY_ARGS)
     def table_exists(self, identifier: Union[str, Identifier]) -> bool:
         """Check if a table exists.

tests/catalog/test_rest.py

@@ -681,6 +681,51 @@ def test_update_namespace_properties_200(rest_mock: Mocker) -> None:
     assert response == PropertiesUpdateSummary(removed=[], updated=["prop"], missing=["abc"])
 
 
+def test_namespace_exists_200(rest_mock: Mocker) -> None:
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/fokko",
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+
+    assert catalog.namespace_exists("fokko")
+
+
+def test_namespace_exists_204(rest_mock: Mocker) -> None:
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/fokko",
+        status_code=204,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+
+    assert catalog.namespace_exists("fokko")
+
+
+def test_namespace_exists_404(rest_mock: Mocker) -> None:
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/fokko",
+        status_code=404,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+
+    assert not catalog.namespace_exists("fokko")
+
+
+def test_namespace_exists_500(rest_mock: Mocker) -> None:
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/fokko",
+        status_code=500,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+
+    with pytest.raises(ServerError):
+        catalog.namespace_exists("fokko")
+
+
 def test_update_namespace_properties_404(rest_mock: Mocker) -> None:
     rest_mock.post(
         f"{TEST_URI}v1/namespaces/fokko/properties",

tests/integration/test_rest_catalog.py

@@ -0,0 +1,63 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# pylint:disable=redefined-outer-name
+
+import pytest
+
+from pyiceberg.catalog.rest import RestCatalog
+
+TEST_NAMESPACE_IDENTIFIER = "TEST NS"
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog")])
+def test_namespace_exists(catalog: RestCatalog) -> None:
+    if not catalog.namespace_exists(TEST_NAMESPACE_IDENTIFIER):
+        catalog.create_namespace(TEST_NAMESPACE_IDENTIFIER)
+
+    assert catalog.namespace_exists(TEST_NAMESPACE_IDENTIFIER)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog")])
+def test_namespace_not_exists(catalog: RestCatalog) -> None:
+    if catalog.namespace_exists(TEST_NAMESPACE_IDENTIFIER):
+        catalog.drop_namespace(TEST_NAMESPACE_IDENTIFIER)
+
+    assert not catalog.namespace_exists(TEST_NAMESPACE_IDENTIFIER)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog")])
+def test_create_namespace_if_not_exists(catalog: RestCatalog) -> None:
+    if catalog.namespace_exists(TEST_NAMESPACE_IDENTIFIER):
+        catalog.drop_namespace(TEST_NAMESPACE_IDENTIFIER)
+
+    catalog.create_namespace_if_not_exists(TEST_NAMESPACE_IDENTIFIER)
+
+    assert catalog.namespace_exists(TEST_NAMESPACE_IDENTIFIER)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog")])
+def test_create_namespace_if_already_existing(catalog: RestCatalog) -> None:
+    if not catalog.namespace_exists(TEST_NAMESPACE_IDENTIFIER):
+        catalog.create_namespace(TEST_NAMESPACE_IDENTIFIER)
+
+    catalog.create_namespace_if_not_exists(TEST_NAMESPACE_IDENTIFIER)
+
+    assert catalog.namespace_exists(TEST_NAMESPACE_IDENTIFIER)