Added qualified_name parameter to AtlanClient.append_terms to allow use of either guid or qualified_name.

Author: ErnestoLoma

pyatlan/client/atlan.py

@@ -313,8 +313,7 @@ def get_asset_by_qualified_name(
             raw_json["entity"]["attributes"].update(
                 raw_json["entity"]["relationshipAttributes"]
             )
-            asset = AssetResponse[A](**raw_json).entity
-            asset.is_incomplete = False
+            asset = self.handle_relationships(raw_json)
             if not isinstance(asset, asset_type):
                 raise NotFoundError(
                     message=f"Asset with qualifiedName {qualified_name} "
@@ -345,16 +344,7 @@ def get_asset_by_guid(
                 GET_ENTITY_BY_GUID.format_path_with_params(guid),
                 query_params,
             )
-            if (
-                "relationshipAttributes" in raw_json["entity"]
-                and raw_json["entity"]["relationshipAttributes"]
-            ):
-                raw_json["entity"]["attributes"].update(
-                    raw_json["entity"]["relationshipAttributes"]
-                )
-            raw_json["entity"]["relationshipAttributes"] = {}
-            asset = AssetResponse[A](**raw_json).entity
-            asset.is_incomplete = False
+            asset = self.handle_relationships(raw_json)
             if not isinstance(asset, asset_type):
                 raise NotFoundError(
                     message=f"Asset with GUID {guid} is not of the type requested: {asset_type.__name__}.",
@@ -366,6 +356,19 @@ def get_asset_by_guid(
                 raise NotFoundError(message=ae.user_message, code=ae.code) from ae
             raise ae
 
+    def handle_relationships(self, raw_json):
+        if (
+            "relationshipAttributes" in raw_json["entity"]
+            and raw_json["entity"]["relationshipAttributes"]
+        ):
+            raw_json["entity"]["attributes"].update(
+                raw_json["entity"]["relationshipAttributes"]
+            )
+        raw_json["entity"]["relationshipAttributes"] = {}
+        asset = AssetResponse[A](**raw_json).entity
+        asset.is_incomplete = False
+        return asset
+
     @validate_arguments()
     def retrieve_minimal(self, guid: str, asset_type: Type[A]) -> A:
         return self.get_asset_by_guid(
@@ -605,9 +608,24 @@ def replace_custom_metadata(self, guid: str, custom_metadata: CustomMetadata):
 
     @validate_arguments()
     def append_terms(
-        self, guid: str, asset_type: Type[A], terms: list[AtlasGlossaryTerm]
+        self,
+        asset_type: Type[A],
+        terms: list[AtlasGlossaryTerm],
+        guid: Optional[str] = None,
+        qualified_name: Optional[str] = None,
     ) -> A:
-        asset = self.get_asset_by_guid(guid=guid, asset_type=asset_type)
+        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")
         if not terms:
             return asset
         replacement_terms: list[AtlasGlossaryTerm] = []

tests/integration/test_client.py

@@ -33,7 +33,7 @@ def database(
 ) -> Generator[Database, None, None]:
 
     database = Database.create(
-        name="Integration_Test_Entity_DB",
+        name=f"Integration_Test_Entity_DB{next(iter_count)}",
         connection_qualified_name=connection.attributes.qualified_name,
     )
     database = client.upsert(database).assets_created(Database)[0]
@@ -75,7 +75,7 @@ def test_register_client():
     assert AtlanClient.get_default_client() == client
 
 
-def test_append_terms(
+def test_append_terms_with_guid(
     client: AtlanClient,
     make_term: Callable[[str], AtlasGlossaryTerm],
     database: Database,
@@ -90,3 +90,20 @@ def test_append_terms(
     database = client.get_asset_by_guid(guid=database.guid, asset_type=Database)
     assert len(database.terms) == 1
     assert database.terms[0].guid == term.guid
+
+
+def test_append_terms_with_qualified_name(
+    client: AtlanClient,
+    make_term: Callable[[str], AtlasGlossaryTerm],
+    database: Database,
+):
+    term = make_term("Term1")
+
+    assert (
+        database := client.append_terms(
+            qualified_name=database.qualified_name, asset_type=Database, terms=[term]
+        )
+    )
+    database = client.get_asset_by_guid(guid=database.guid, asset_type=Database)
+    assert len(database.terms) == 1
+    assert database.terms[0].guid == term.guid

tests/unit/test_client.py

@@ -9,33 +9,41 @@
 
 
 @pytest.mark.parametrize(
-    "guid, asset_type, terms, message",
+    "guid, qualified_name, asset_type, terms, message",
     [
         (
             "123",
+            None,
             Table,
             None,
-            "1 validation error for AppendTerms\\nterms\\n  none is not an allowed value "
-            "\(type=type_error.none.not_allowed\)",  # noqa: W605
+            "1 validation error for AppendTerms\\nterms\\n  none is not an allowed value ",
         ),
         (
+            None,
             None,
             Table,
             [AtlasGlossaryTerm()],
-            "1 validation error for AppendTerms\\nguid\\n  none is not an allowed value "
-            "\(type=type_error.none.not_allowed\)",  # noqa: W605
+            "Either guid or qualified name must be specified",
         ),
         (
             "123",
             None,
+            None,
             [AtlasGlossaryTerm()],
-            "1 validation error for AppendTerms\\nasset_type\\n  none is not an allowed value "
-            "\(type=type_error.none.not_allowed\)",  # noqa: W605
+            "1 validation error for AppendTerms\\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",
         ),
     ],
 )
 def test_append_terms_with_invalid_parameter_raises_valueerror(
     guid,
+    qualified_name,
     asset_type,
     terms,
     message,
@@ -46,7 +54,9 @@ def test_append_terms_with_invalid_parameter_raises_valueerror(
     client = AtlanClient()
 
     with pytest.raises(ValueError, match=message):
-        client.append_terms(guid=guid, asset_type=asset_type, terms=terms)
+        client.append_terms(
+            guid=guid, qualified_name=qualified_name, asset_type=asset_type, terms=terms
+        )
 
 
 def test_append_with_valid_guid_and_no_terms_returns_asset(monkeypatch):