Enable exception suppression during namespace creation

ndrluis · ndrluis · commit 26e55be721ac · 2024-05-10T19:02:38.000-03:00
diff --git a/pyiceberg/catalog/__init__.py b/pyiceberg/catalog/__init__.py
@@ -466,15 +466,21 @@ def _commit_table(self, table_request: CommitTableRequest) -> CommitTableRespons
         """
 
     @abstractmethod
-    def create_namespace(self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT) -> None:
+    def create_namespace(
+        self,
+        namespace: Union[str, Identifier],
+        properties: Properties = EMPTY_DICT,
+        error_if_exists: bool = True,
+    ) -> None:
         """Create a namespace in the catalog.
 
         Args:
             namespace (str | Identifier): Namespace identifier.
             properties (Properties): A string dictionary of properties for the given namespace.
+            error_if_exists (bool): If True, raise an error when the namespace already exists. Default is True.
 
         Raises:
-            NamespaceAlreadyExistsError: If a namespace with the given name already exists.
+            NamespaceAlreadyExistsError: If a namespace with the given name already exists and error_if_exists is True.
         """
 
     @abstractmethod
diff --git a/pyiceberg/catalog/dynamodb.py b/pyiceberg/catalog/dynamodb.py
@@ -319,16 +319,22 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
 
         return self.load_table(to_identifier)
 
-    def create_namespace(self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT) -> None:
+    def create_namespace(
+        self,
+        namespace: Union[str, Identifier],
+        properties: Properties = EMPTY_DICT,
+        error_if_exists: bool = True,
+    ) -> None:
         """Create a namespace in the catalog.
 
         Args:
-            namespace: Namespace identifier.
-            properties: A string dictionary of properties for the given namespace.
+            namespace (str | Identifier): Namespace identifier.
+            properties (Properties): A string dictionary of properties for the given namespace.
+            error_if_exists (bool): If True, raise an error when the namespace already exists. Default is True.
 
         Raises:
             ValueError: If the identifier is invalid.
-            AlreadyExistsError: If a namespace with the given name already exists.
+            NamespaceAlreadyExistsError: If a namespace with the given name already exists and error_if_exists is True.
         """
         database_name = self.identifier_to_database(namespace)
 
@@ -338,6 +344,8 @@ def create_namespace(self, namespace: Union[str, Identifier], properties: Proper
                 condition_expression=f"attribute_not_exists({DYNAMODB_COL_NAMESPACE})",
             )
         except ConditionalCheckFailedException as e:
+            if not error_if_exists:
+                return
             raise NamespaceAlreadyExistsError(f"Database {database_name} already exists") from e
 
     def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
diff --git a/pyiceberg/catalog/glue.py b/pyiceberg/catalog/glue.py
@@ -593,21 +593,29 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
 
         return self.load_table(to_identifier)
 
-    def create_namespace(self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT) -> None:
+    def create_namespace(
+        self,
+        namespace: Union[str, Identifier],
+        properties: Properties = EMPTY_DICT,
+        error_if_exists: bool = True,
+    ) -> None:
         """Create a namespace in the catalog.
 
         Args:
-            namespace: Namespace identifier.
-            properties: A string dictionary of properties for the given namespace.
+            namespace (str | Identifier): Namespace identifier.
+            properties (Properties): A string dictionary of properties for the given namespace.
+            error_if_exists (bool): If True, raise an error when the namespace already exists. Default is True.
 
         Raises:
             ValueError: If the identifier is invalid.
-            AlreadyExistsError: If a namespace with the given name already exists.
+            NamespaceAlreadyExistsError: If a namespace with the given name already exists and error_if_exists is True.
         """
         database_name = self.identifier_to_database(namespace)
         try:
             self.glue.create_database(DatabaseInput=_construct_database_input(database_name, properties))
         except self.glue.exceptions.AlreadyExistsException as e:
+            if not error_if_exists:
+                return
             raise NamespaceAlreadyExistsError(f"Database {database_name} already exists") from e
 
     def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
diff --git a/pyiceberg/catalog/hive.py b/pyiceberg/catalog/hive.py
@@ -489,16 +489,22 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
             raise NoSuchNamespaceError(f"Database does not exists: {to_database_name}") from e
         return self.load_table(to_identifier)
 
-    def create_namespace(self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT) -> None:
+    def create_namespace(
+        self,
+        namespace: Union[str, Identifier],
+        properties: Properties = EMPTY_DICT,
+        error_if_exists: bool = True,
+    ) -> None:
         """Create a namespace in the catalog.
 
         Args:
             namespace: Namespace identifier.
             properties: A string dictionary of properties for the given namespace.
+            error_if_exists (bool): If True, raise an error when the namespace already exists. Default is True.
 
         Raises:
             ValueError: If the identifier is invalid.
-            AlreadyExistsError: If a namespace with the given name already exists.
+            NamespaceAlreadyExistsError: If a namespace with the given name already exists and error_if_exist is True.
         """
         database_name = self.identifier_to_database(namespace)
         hive_database = HiveDatabase(name=database_name, parameters=properties)
@@ -507,6 +513,8 @@ def create_namespace(self, namespace: Union[str, Identifier], properties: Proper
             with self._client as open_client:
                 open_client.create_database(_annotate_namespace(hive_database, properties))
         except AlreadyExistsException as e:
+            if not error_if_exists:
+                return
             raise NamespaceAlreadyExistsError(f"Database {database_name} already exists") from e
 
     def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
diff --git a/pyiceberg/catalog/noop.py b/pyiceberg/catalog/noop.py
@@ -94,7 +94,9 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
     def _commit_table(self, table_request: CommitTableRequest) -> CommitTableResponse:
         raise NotImplementedError
 
-    def create_namespace(self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT) -> None:
+    def create_namespace(
+        self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT, error_if_exists: bool = True
+    ) -> None:
         raise NotImplementedError
 
     def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
diff --git a/pyiceberg/catalog/rest.py b/pyiceberg/catalog/rest.py
@@ -708,13 +708,30 @@ def _commit_table(self, table_request: CommitTableRequest) -> CommitTableRespons
         return CommitTableResponse(**response.json())
 
     @retry(**_RETRY_ARGS)
-    def create_namespace(self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT) -> None:
+    def create_namespace(
+        self,
+        namespace: Union[str, Identifier],
+        properties: Properties = EMPTY_DICT,
+        error_if_exists: bool = True,
+    ) -> None:
+        """Create a namespace in the catalog.
+
+        Args:
+            namespace (str | Identifier): Namespace identifier.
+            properties (Properties): A string dictionary of properties for the given namespace.
+            error_if_exists (bool): If True, raise an error when the namespace already exists. Default is True.
+
+        Raises:
+            NamespaceAlreadyExistsError: If a namespace with the given name already exists and error_if_exists is True.
+        """
         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:
+            if exc.response.status_code == 409 and not error_if_exists:
+                return
             self._handle_non_200_response(exc, {404: NoSuchNamespaceError, 409: NamespaceAlreadyExistsError})
 
     @retry(**_RETRY_ARGS)
diff --git a/pyiceberg/catalog/sql.py b/pyiceberg/catalog/sql.py
@@ -452,20 +452,28 @@ def _namespace_exists(self, identifier: Union[str, Identifier]) -> bool:
                 return True
         return False
 
-    def create_namespace(self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT) -> None:
+    def create_namespace(
+        self,
+        namespace: Union[str, Identifier],
+        properties: Properties = EMPTY_DICT,
+        error_if_exists: bool = True,
+    ) -> None:
         """Create a namespace in the catalog.
 
         Args:
             namespace (str | Identifier): Namespace identifier.
             properties (Properties): A string dictionary of properties for the given namespace.
+            error_if_exists (bool): If True, raise an error when the namespace already exists. Default is True.
 
         Raises:
-            NamespaceAlreadyExistsError: If a namespace with the given name already exists.
+            NamespaceAlreadyExistsError: If a namespace with the given name already exists and error_if_exists is True.
         """
         if not properties:
             properties = IcebergNamespaceProperties.NAMESPACE_MINIMAL_PROPERTIES
         database_name = self.identifier_to_database(namespace)
         if self._namespace_exists(database_name):
+            if not error_if_exists:
+                return
             raise NamespaceAlreadyExistsError(f"Database {database_name} already exists")
 
         create_properties = properties if properties else IcebergNamespaceProperties.NAMESPACE_MINIMAL_PROPERTIES
diff --git a/tests/catalog/integration_test_dynamodb.py b/tests/catalog/integration_test_dynamodb.py
@@ -184,6 +184,11 @@ def test_create_duplicate_namespace(test_catalog: Catalog, database_name: str) -
         test_catalog.create_namespace(database_name)
 
 
+def test_create_duplicate_namespace_when_error_if_exists_is_false(test_catalog: Catalog, database_name: str) -> None:
+    test_catalog.create_namespace(database_name)
+    test_catalog.create_namespace(database_name, error_if_exists=False)
+
+
 def test_create_namespace_with_comment_and_location(test_catalog: Catalog, database_name: str) -> None:
     test_location = get_s3_path(get_bucket_name(), database_name)
     test_properties = {
diff --git a/tests/catalog/integration_test_glue.py b/tests/catalog/integration_test_glue.py
@@ -291,6 +291,11 @@ def test_create_duplicate_namespace(test_catalog: Catalog, database_name: str) -
         test_catalog.create_namespace(database_name)
 
 
+def test_create_duplicate_namespace_when_error_if_exists_is_false(test_catalog: Catalog, database_name: str) -> None:
+    test_catalog.create_namespace(database_name)
+    test_catalog.create_namespace(database_name, error_if_exists=False)
+
+
 def test_create_namespace_with_comment_and_location(test_catalog: Catalog, database_name: str) -> None:
     test_location = get_s3_path(get_bucket_name(), database_name)
     test_properties = {
diff --git a/tests/catalog/test_base.py b/tests/catalog/test_base.py
@@ -195,9 +195,13 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
         )
         return self.__tables[to_identifier]
 
-    def create_namespace(self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT) -> None:
+    def create_namespace(
+        self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT, error_if_exists: bool = True
+    ) -> None:
         namespace = Catalog.identifier_to_tuple(namespace)
         if namespace in self.__namespaces:
+            if not error_if_exists:
+                return
             raise NamespaceAlreadyExistsError(f"Namespace already exists: {namespace}")
         else:
             self.__namespaces[namespace] = properties if properties else {}
@@ -541,6 +545,15 @@ def test_create_namespace_raises_error_on_existing_namespace(catalog: InMemoryCa
         catalog.create_namespace(TEST_TABLE_NAMESPACE, TEST_TABLE_PROPERTIES)
 
 
+def test_create_namespace_does_not_raises_error_on_existing_namespace_when_error_if_exists_is_false(
+    catalog: InMemoryCatalog,
+) -> None:
+    # Given
+    catalog.create_namespace(TEST_TABLE_NAMESPACE, TEST_TABLE_PROPERTIES)
+    # When
+    catalog.create_namespace(TEST_TABLE_NAMESPACE, TEST_TABLE_PROPERTIES, error_if_exists=False)
+
+
 def test_get_namespace_metadata_raises_error_when_namespace_does_not_exist(catalog: InMemoryCatalog) -> None:
     with pytest.raises(NoSuchNamespaceError, match=NO_SUCH_NAMESPACE_ERROR):
         catalog.load_namespace_properties(TEST_TABLE_NAMESPACE)
diff --git a/tests/catalog/test_dynamodb.py b/tests/catalog/test_dynamodb.py
@@ -448,6 +448,16 @@ def test_create_duplicated_namespace(_bucket_initialize: None, database_name: st
         test_catalog.create_namespace(namespace=database_name, properties={"test": "test"})
 
 
+@mock_aws
+def test_create_duplicated_namespace_when_error_if_exists_is_false(_bucket_initialize: None, database_name: str) -> None:
+    test_catalog = DynamoDbCatalog("test_ddb_catalog")
+    test_catalog.create_namespace(namespace=database_name)
+    loaded_database_list = test_catalog.list_namespaces()
+    assert len(loaded_database_list) == 1
+    assert (database_name,) in loaded_database_list
+    test_catalog.create_namespace(namespace=database_name, properties={"test": "test"}, error_if_exists=False)
+
+
 @mock_aws
 def test_drop_namespace(_bucket_initialize: None, database_name: str) -> None:
     test_catalog = DynamoDbCatalog("test_ddb_catalog")
diff --git a/tests/catalog/test_glue.py b/tests/catalog/test_glue.py
@@ -498,6 +498,18 @@ def test_create_duplicated_namespace(_bucket_initialize: None, moto_endpoint_url
         test_catalog.create_namespace(namespace=database_name, properties={"test": "test"})
 
 
+@mock_aws
+def test_create_duplicated_namespace_when_error_if_exists_is_false(
+    _bucket_initialize: None, moto_endpoint_url: str, database_name: str
+) -> None:
+    test_catalog = GlueCatalog("glue", **{"s3.endpoint": moto_endpoint_url})
+    test_catalog.create_namespace(namespace=database_name)
+    loaded_database_list = test_catalog.list_namespaces()
+    assert len(loaded_database_list) == 1
+    assert (database_name,) in loaded_database_list
+    test_catalog.create_namespace(namespace=database_name, properties={"test": "test"}, error_if_exists=False)
+
+
 @mock_aws
 def test_drop_namespace(_bucket_initialize: None, moto_endpoint_url: str, database_name: str) -> None:
     test_catalog = GlueCatalog("glue", **{"s3.endpoint": moto_endpoint_url})
diff --git a/tests/catalog/test_hive.py b/tests/catalog/test_hive.py
@@ -996,7 +996,7 @@ def test_create_database() -> None:
     )
 
 
-def test_create_database_already_exists() -> None:
+def test_create_namespace_already_exists() -> None:
     catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
 
     catalog._client = MagicMock()
@@ -1008,6 +1008,15 @@ def test_create_database_already_exists() -> None:
     assert "Database default already exists" in str(exc_info.value)
 
 
+def test_create_namespace_already_exists_when_error_if_exists_is_false() -> None:
+    catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
+
+    catalog._client = MagicMock()
+    catalog._client.__enter__().create_database.side_effect = AlreadyExistsException(message="Database default already exists")
+
+    catalog.create_namespace("default", error_if_exists=False)
+
+
 def test_load_namespace_properties(hive_database: HiveDatabase) -> None:
     catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
 
diff --git a/tests/catalog/test_rest.py b/tests/catalog/test_rest.py
@@ -519,6 +519,24 @@ def test_create_namespace_409(rest_mock: Mocker) -> None:
     assert "Namespace already exists" in str(e.value)
 
 
+def test_create_namespace_409_with_error_if_exists_is_false(rest_mock: Mocker) -> None:
+    namespace = "examples"
+    rest_mock.post(
+        f"{TEST_URI}v1/namespaces",
+        json={
+            "error": {
+                "message": "Namespace already exists: fokko in warehouse 8bcb0838-50fc-472d-9ddb-8feb89ef5f1e",
+                "type": "AlreadyExistsException",
+                "code": 409,
+            }
+        },
+        status_code=409,
+        request_headers=TEST_HEADERS,
+    )
+
+    RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).create_namespace(namespace, error_if_exists=False)
+
+
 def test_drop_namespace_404(rest_mock: Mocker) -> None:
     namespace = "examples"
     rest_mock.delete(
diff --git a/tests/catalog/test_sql.py b/tests/catalog/test_sql.py
@@ -646,6 +646,18 @@ def test_create_duplicate_namespace(catalog: SqlCatalog, database_name: str) ->
         catalog.create_namespace(database_name)
 
 
+@pytest.mark.parametrize(
+    'catalog',
+    [
+        lazy_fixture('catalog_memory'),
+        lazy_fixture('catalog_sqlite'),
+    ],
+)
+def test_create_namespace_if_not_exists(catalog: SqlCatalog, database_name: str) -> None:
+    catalog.create_namespace(database_name)
+    catalog.create_namespace(database_name, error_if_exists=False)
+
+
 @pytest.mark.parametrize(
     'catalog',
     [
