Add remove_terms method to AtlanClient.

Author: ErnestoLoma

pyatlan/client/atlan.py

@@ -665,3 +665,40 @@ def replace_terms(
         if assets := response.assets_updated(asset_type=asset_type):
             return assets[0]
         return asset
+
+    @validate_arguments()
+    def remove_terms(
+        self,
+        asset_type: Type[A],
+        terms: list[AtlasGlossaryTerm],
+        guid: Optional[str] = None,
+        qualified_name: Optional[str] = None,
+    ) -> A:
+        if not terms:
+            raise ValueError("A list of terms to remove must be specified")
+        if guid:
+            if qualified_name:
+                raise ValueError(
+                    "Either guid or qualified_name can be be specified not both"
+                )
+            asset = self.get_asset_by_guid(guid=guid, asset_type=asset_type)
+        elif qualified_name:
+            asset = self.get_asset_by_qualified_name(
+                qualified_name=qualified_name, asset_type=asset_type
+            )
+        else:
+            raise ValueError("Either guid or qualified name must be specified")
+        replacement_terms: list[AtlasGlossaryTerm] = []
+        guids_to_be_removed = {t.guid for t in terms}
+        if existing_terms := asset.terms:
+            replacement_terms.extend(
+                term
+                for term in existing_terms
+                if term.relationship_status != "DELETED"
+                and term.guid not in guids_to_be_removed
+            )
+        asset.terms = replacement_terms
+        response = self.upsert(entity=asset)
+        if assets := response.assets_updated(asset_type=asset_type):
+            return assets[0]
+        return asset

tests/integration/test_client.py

@@ -184,3 +184,38 @@ def test_replace_all_term(
     deleted_terms = [t for t in database.terms if t.relationship_status == "DELETED"]
     assert len(deleted_terms) == 1
     assert deleted_terms[0].guid == original_term.guid
+
+
+def test_remove_term(
+    client: AtlanClient,
+    make_term: Callable[[str], AtlasGlossaryTerm],
+    database: Database,
+):
+    original_term = make_term("Term1")
+    another_term = make_term("Term2")
+    assert (
+        database := client.append_terms(
+            qualified_name=database.qualified_name,
+            asset_type=Database,
+            terms=[
+                AtlasGlossaryTerm.ref_by_guid(guid=original_term.guid),
+                AtlasGlossaryTerm.ref_by_guid(guid=another_term.guid),
+            ],
+        )
+    )
+
+    assert (
+        database := client.remove_terms(
+            guid=database.guid,
+            asset_type=Database,
+            terms=[AtlasGlossaryTerm.ref_by_guid(original_term.guid)],
+        )
+    )
+
+    database = client.get_asset_by_guid(guid=database.guid, asset_type=Database)
+    assert len(database.terms) == 2
+    deleted_terms = [t for t in database.terms if t.relationship_status == "DELETED"]
+    assert len(deleted_terms) == 1
+    assert deleted_terms[0].guid == original_term.guid
+    active_terms = [t for t in database.terms if t.relationship_status != "DELETED"]
+    assert active_terms[0].guid == another_term.guid

tests/unit/test_client.py

@@ -225,3 +225,91 @@ def test_replace_terms(
             asset := client.replace_terms(guid=guid, asset_type=asset_type, terms=terms)
         )
         assert asset.terms == terms
+
+
+@pytest.mark.parametrize(
+    "guid, qualified_name, asset_type, terms, message",
+    [
+        (
+            None,
+            None,
+            Table,
+            [AtlasGlossaryTerm()],
+            "Either guid or qualified name must be specified",
+        ),
+        (
+            "123",
+            None,
+            None,
+            [AtlasGlossaryTerm()],
+            "1 validation error for RemoveTerms\\nasset_type\\n  none is not an allowed value ",
+        ),
+        (
+            "123",
+            "default/abc",
+            Table,
+            [AtlasGlossaryTerm()],
+            "Either guid or qualified_name can be be specified not both",
+        ),
+        (
+            "123",
+            None,
+            Table,
+            None,
+            "1 validation error for RemoveTerms\\nterms\\n  none is not an allowed value ",
+        ),
+        (
+            "123",
+            None,
+            Table,
+            [],
+            "A list of terms to remove must be specified",
+        ),
+    ],
+)
+def test_remove_terms_with_invalid_parameter_raises_valueerror(
+    guid,
+    qualified_name,
+    asset_type,
+    terms,
+    message,
+    monkeypatch,
+):
+    monkeypatch.setenv("ATLAN_BASE_URL", "https://name.atlan.com")
+    monkeypatch.setenv("ATLAN_API_KEY", "abkj")
+    client = AtlanClient()
+
+    with pytest.raises(ValueError, match=message):
+        client.remove_terms(
+            guid=guid, qualified_name=qualified_name, asset_type=asset_type, terms=terms
+        )
+
+
+def test_remove_with_valid_guid_when_terms_present_returns_asset_with_terms_removed(
+    monkeypatch,
+):
+    monkeypatch.setenv("ATLAN_BASE_URL", "https://name.atlan.com")
+    monkeypatch.setenv("ATLAN_API_KEY", "abkj")
+    asset_type = Table
+    with patch.multiple(
+        AtlanClient, get_asset_by_guid=DEFAULT, upsert=DEFAULT
+    ) as mock_methods:
+        table = Table(attributes=Table.Attributes())
+        exisiting_term = AtlasGlossaryTerm()
+        exisiting_term.guid = "b4113341-251b-4adc-81fb-2420501c30e6"
+        other_term = AtlasGlossaryTerm()
+        other_term.guid = "b267858d-8316-4c41-a56a-6e9b840cef4a"
+        table.attributes.meanings = [exisiting_term, other_term]
+        mock_methods["get_asset_by_guid"].return_value = table
+        mock_methods["upsert"].return_value.assets_updated.return_value = [table]
+        client = AtlanClient()
+        guid = "123"
+
+        assert (
+            asset := client.remove_terms(
+                guid=guid, asset_type=asset_type, terms=[exisiting_term]
+            )
+        )
+        assert (updated_terms := asset.terms)
+        assert len(updated_terms) == 1
+        assert other_term in updated_terms