Merge pull request #674 from atlanhq/APP-6910

[test] Fixed `suggestions_test` and added `delete_type` param to `purge_by_guid()` method

Author: Aryamanz29

pyatlan/client/asset.py

@@ -786,21 +786,32 @@ def update_replacing_cm(
         )
 
     @validate_arguments
-    def purge_by_guid(self, guid: Union[str, List[str]]) -> AssetMutationResponse:
+    def purge_by_guid(
+        self,
+        guid: Union[str, List[str]],
+        delete_type: AtlanDeleteType = AtlanDeleteType.PURGE,
+    ) -> AssetMutationResponse:
         """
-        Hard-deletes (purges) one or more assets by their unique identifier (GUID).
-        This operation is irreversible.
+        Deletes one or more assets by their unique identifier (GUID) using the specified delete type.
+
+        :param guid: unique identifier(s) (GUIDs) of one or more assets to delete
+        :param delete_type: type of deletion to perform:
 
-        :param guid: unique identifier(s) (GUIDs) of one or more assets to hard-delete
-        :returns: details of the hard-deleted asset(s)
+            - PURGE: completely removes entity and all audit/history traces (default, irreversible)
+            - HARD: physically removes entity but keeps audit history (irreversible)
+
+        :returns: details of the deleted asset(s)
         :raises AtlanError: on any API communication issue
+
+        .. warning::
+            PURGE and HARD deletions are irreversible operations. Use with caution.
         """
         guids: List[str] = []
         if isinstance(guid, list):
             guids.extend(guid)
         else:
             guids.append(guid)
-        query_params = {"deleteType": AtlanDeleteType.PURGE.value, "guid": guids}
+        query_params = {"deleteType": delete_type.value, "guid": guids}
         raw_json = self._client._call_api(
             DELETE_ENTITIES_BY_GUIDS, query_params=query_params
         )

tests/integration/client.py

@@ -6,6 +6,7 @@
 import pytest
 
 from pyatlan.client.atlan import AtlanClient
+from pyatlan.model.enums import AtlanDeleteType
 from pyatlan.model.response import A
 
 LOGGER = logging.getLogger(__name__)
@@ -31,9 +32,14 @@ def client() -> Generator[AtlanClient, None, None]:
     yield client
 
 
-def delete_asset(client: AtlanClient, asset_type: Type[A], guid: str) -> None:
+def delete_asset(
+    client: AtlanClient,
+    asset_type: Type[A],
+    guid: str,
+    delete_type: AtlanDeleteType = AtlanDeleteType.PURGE,
+) -> None:
     # These assertions check the cleanup actually worked
-    r = client.asset.purge_by_guid(guid)
+    r = client.asset.purge_by_guid(guid, delete_type)
     s = r is not None
     s = s and len(r.assets_deleted(asset_type)) == 1
     s = s and r.assets_deleted(asset_type)[0].guid == guid

tests/integration/suggestions_test.py

@@ -2,6 +2,7 @@
 from typing import Callable, Generator
 
 import pytest
+from pydantic import StrictStr
 
 from pyatlan.client.atlan import AtlanClient
 from pyatlan.model.assets import (
@@ -16,7 +17,7 @@
     View,
 )
 from pyatlan.model.core import AtlanTag, AtlanTagName
-from pyatlan.model.enums import AtlanConnectorType
+from pyatlan.model.enums import AtlanConnectorType, AtlanDeleteType
 from pyatlan.model.group import AtlanGroup
 from pyatlan.model.response import AssetMutationResponse
 from pyatlan.model.suggestions import Suggestions
@@ -47,6 +48,23 @@
 TERM_NAME2 = PREFIX + "term2"
 
 
+def create_glossary(client: AtlanClient, name: str) -> AtlasGlossary:
+    g = AtlasGlossary.create(name=StrictStr(name))
+    r = client.asset.save(g)
+    return r.assets_created(AtlasGlossary)[0]
+
+
+@pytest.fixture(scope="module")
+def glossary(
+    client: AtlanClient,
+) -> Generator[AtlasGlossary, None, None]:
+    g = create_glossary(client, PREFIX)
+    yield g
+    delete_asset(
+        client, guid=g.guid, asset_type=AtlasGlossary, delete_type=AtlanDeleteType.HARD
+    )
+
+
 @pytest.fixture(scope="module")
 def wait_for_consistency():
     """
@@ -383,7 +401,12 @@ def term1(
         qualified_name=t1c1.qualified_name,
         terms=[AtlasGlossaryTerm.ref_by_guid(t.guid)],
     )
-    delete_asset(client, guid=t.guid, asset_type=AtlasGlossaryTerm)
+    delete_asset(
+        client,
+        guid=t.guid,
+        asset_type=AtlasGlossaryTerm,
+        delete_type=AtlanDeleteType.HARD,
+    )
 
 
 @pytest.fixture(scope="module")
@@ -427,7 +450,12 @@ def term2(
         qualified_name=v1c1.qualified_name,
         terms=[AtlasGlossaryTerm.ref_by_guid(t.guid)],
     )
-    delete_asset(client, guid=t.guid, asset_type=AtlasGlossaryTerm)
+    delete_asset(
+        client,
+        guid=t.guid,
+        asset_type=AtlasGlossaryTerm,
+        delete_type=AtlanDeleteType.HARD,
+    )
 
 
 @pytest.fixture(scope="module")
@@ -485,7 +513,9 @@ def table3(
     assert (tables := response.assets_created(asset_type=Table))
     assert len(tables) == 1
     yield tables[0]
-    # delete_asset(client, guid=tables[0].guid, asset_type=Table)
+    delete_asset(
+        client, guid=tables[0].guid, asset_type=Table, delete_type=AtlanDeleteType.HARD
+    )
 
 
 def test_tables(