[chores + tests] Removed commented code and refactored tests

Author: Aryamanz29

pyatlan/client/atlan.py

@@ -646,7 +646,7 @@ def _create_params(
             params["params"] = query_params
         if request_obj is not None:
             if isinstance(request_obj, AtlanObject):
-                params["data"] = AtlanRequest(request_obj, client=self).json()
+                params["data"] = AtlanRequest(instance=request_obj, client=self).json()
             elif api.consumes == APPLICATION_ENCODED_FORM:
                 params["data"] = request_obj
             else:

pyatlan/model/core.py

@@ -43,20 +43,7 @@ def __new__(cls, *args, **kwargs):
         return obj
 
     def __init__(self, display_text: str):
-        # from pyatlan.client.atlan import AtlanClient
-
-        # if not (
-        #     id := AtlanClient.get_current_client().atlan_tag_cache.get_id_for_name(
-        #         display_text
-        #     )
-        # ):
-        #     raise ValueError(f"{display_text} is not a valid Classification")
         self._display_text = display_text
-        # self._id = id
-
-    # @property
-    # def id(self):
-    #     return self._id
 
     @classmethod
     def get_deleted_sentinel(cls) -> "AtlanTagName":
@@ -84,35 +71,12 @@ def __eq__(self, other):
             and self._display_text == other._display_text
         )
 
-    # @classmethod
-    # def _convert_to_display_text(cls, data):
-    #     from pyatlan.client.atlan import AtlanClient
-
-    #     if isinstance(data, AtlanTagName):
-    #         return data
-
-    #     if (
-    #         display_text
-    #         := AtlanClient.get_current_client().atlan_tag_cache.get_name_for_id(data)
-    #     ):
-    #         return AtlanTagName(display_text)
-    #     else:
-    #         return cls.get_deleted_sentinel()
-
     @classmethod
     def _convert_to_tag_name(cls, data):
         if isinstance(data, AtlanTagName):
             return data
         return AtlanTagName(data) if data else cls.get_deleted_sentinel()
 
-    # @staticmethod
-    # def json_encode_atlan_tag(atlan_tag_name: "AtlanTagName"):
-    #     from pyatlan.client.atlan import AtlanClient
-
-    #     return AtlanClient.get_current_client().atlan_tag_cache.get_id_for_name(
-    #         atlan_tag_name._display_text
-    #     )
-
 
 class AtlanObject(BaseModel):
     __atlan_extra__: Dict[str, Any] = Field(
@@ -222,20 +186,20 @@ def to_dict(self) -> Union[Dict[str, Any], List[Any], Any]:
 
 
 class AtlanRequest:
-    def __init__(self, model: AtlanObject, client: AtlanClient):
+    def __init__(self, instance: AtlanObject, client: AtlanClient):
         self.client = client
-        self.model = model
+        self.instance = instance
         self.retranslators = [
             AtlanTagRetranslator(client),
             # add others...
         ]
-        # Do: model.json() → parse → translate → store
+        # Do: instance.json() → parse → translate → store
         try:
-            raw_json = self.model.json(
+            raw_json = self.instance.json(
                 by_alias=True, exclude_unset=True, client=self.client
             )
         except TypeError:
-            raw_json = self.model.json(
+            raw_json = self.instance.json(
                 by_alias=True,
                 exclude_unset=True,
             )
@@ -328,7 +292,7 @@ class Config:
     )
 
     attributes: Optional[Dict[str, Any]] = None
-    tag_id: Optional[str] = Field(default=True, exclude=True)
+    tag_id: Optional[str] = Field(default=None, exclude=True)
 
     # @property
     # def source_tag_attachements(self) -> List[SourceTagAttachment]:

pyatlan/model/retranslators.py

@@ -19,8 +19,8 @@ def retranslate(self, data: Dict[str, Any]) -> Dict[str, Any]:
 
 class AtlanTagRetranslator(BaseRetranslator):
     _TYPE_NAME = "typeName"
-    _CLASSIFICATION_NAMES = "classificationNames"
     _SOURCE_ATTACHMENTS = "source_tag_attachements"
+    _CLASSIFICATION_NAMES = {"classificationNames", "purposeClassifications"}
     _CLASSIFICATION_KEYS = {
         "classifications",
         "addOrUpdateClassifications",
@@ -31,34 +31,35 @@ def __init__(self, client: "AtlanClient"):
         self.client = client
 
     def applies_to(self, data: Dict[str, Any]) -> bool:
-        return self._CLASSIFICATION_NAMES in data or any(
+        return any(key in data for key in self._CLASSIFICATION_NAMES) or any(
             key in data for key in self._CLASSIFICATION_KEYS
         )
 
     def retranslate(self, data: Dict[str, Any]) -> Dict[str, Any]:
+        from pyatlan.model.constants import DELETED_
+
         data = data.copy()
 
-        # Convert classification names → IDs
-        if self._CLASSIFICATION_NAMES in data:
-            data[self._CLASSIFICATION_NAMES] = [
-                self.client.atlan_tag_cache.get_id_for_name(str(name))
-                for name in data[self._CLASSIFICATION_NAMES]
-            ]
+        # Convert classification human-readable name → hash ID
+        for key in self._CLASSIFICATION_NAMES:
+            if key in data:
+                data[key] = [
+                    self.client.atlan_tag_cache.get_id_for_name(str(name)) or DELETED_
+                    for name in data[key]
+                ]
 
-        # Convert classification objects
+        # Convert classification objects human-readable name typeName → hash ID
         for key in self._CLASSIFICATION_KEYS:
             if key in data:
                 for classification in data[key]:
                     tag_name = str(classification.get(self._TYPE_NAME))
                     if tag_name:
                         tag_id = self.client.atlan_tag_cache.get_id_for_name(tag_name)
-                        if not tag_id:
-                            continue
-                        classification[self._TYPE_NAME] = tag_id
+                        classification[self._TYPE_NAME] = tag_id if tag_id else DELETED_
 
                         # Rebuild source tag attributes
                         attachments = classification.pop(self._SOURCE_ATTACHMENTS, None)
-                        if attachments:
+                        if attachments and tag_id:
                             attr_id = (
                                 self.client.atlan_tag_cache.get_source_tags_attr_id(
                                     tag_id
@@ -72,5 +73,4 @@ def retranslate(self, data: Dict[str, Any]) -> Dict[str, Any]:
                                     }
                                     for attachment in attachments
                                 ]
-
         return data

pyatlan/model/translators.py

@@ -20,46 +20,60 @@ def translate(self, data: Dict[str, Any]) -> Dict[str, Any]:
 
 
 class AtlanTagTranslator(BaseTranslator):
+    _TAG_ID = "tag_id"
     _TYPE_NAME = "typeName"
-    _CLASSIFICATIONS = "classifications"
-    _CLASSIFICATION_NAMES = "classificationNames"
+    _SOURCE_ATTACHMENTS = "source_tag_attachements"
+    _CLASSIFICATION_NAMES = {"classificationNames", "purposeClassifications"}
+    _CLASSIFICATION_KEYS = {
+        "classifications",
+        "addOrUpdateClassifications",
+        "removeClassifications",
+    }
 
     def __init__(self, client: AtlanClient):
         self.client = client
 
     def applies_to(self, data: Dict[str, Any]) -> bool:
-        return self._CLASSIFICATION_NAMES in data or self._CLASSIFICATIONS in data
+        return any(key in data for key in self._CLASSIFICATION_NAMES) or any(
+            key in data for key in self._CLASSIFICATION_KEYS
+        )
 
     def translate(self, data: Dict[str, Any]) -> Dict[str, Any]:
+        from pyatlan.model.constants import DELETED_
+
         raw_json = data.copy()
 
-        if self._CLASSIFICATION_NAMES in raw_json:
-            raw_json[self._CLASSIFICATION_NAMES] = [
-                self.client.atlan_tag_cache.get_name_for_id(tag_id)
-                for tag_id in raw_json[self._CLASSIFICATION_NAMES]
-            ]
+        # Convert classification hash ID → human-readable name
+        for key in self._CLASSIFICATION_NAMES:
+            if key in raw_json:
+                raw_json[key] = [
+                    self.client.atlan_tag_cache.get_name_for_id(tag_id) or DELETED_
+                    for tag_id in raw_json[key]
+                ]
 
-        if self._CLASSIFICATIONS in raw_json:
-            for classification in raw_json[self._CLASSIFICATIONS]:
-                tag_id = classification.get(self._TYPE_NAME)
-                if tag_id:
-                    tag_name = self.client.atlan_tag_cache.get_name_for_id(tag_id)
-                    if not tag_name:
-                        continue
-                    classification[self._TYPE_NAME] = tag_name
-                    classification["tag_id"] = tag_id
-                    # Check if the tag is a source tag (in that case tag has "attributes")
-                    attr_id = self.client.atlan_tag_cache.get_source_tags_attr_id(
-                        tag_id
-                    )
-                    if attr_id:
-                        classification["source_tag_attachements"] = [
-                            SourceTagAttachment(**source_tag["attributes"])
-                            for source_tag in classification.get("attributes").get(
-                                attr_id
-                            )
-                            if isinstance(source_tag, dict)
-                            and source_tag.get("attributes")
-                        ]
+        # Convert classification objects typeName hash ID → human-readable name
+        for key in self._CLASSIFICATION_KEYS:
+            if key in raw_json:
+                for classification in raw_json[key]:
+                    tag_id = classification.get(self._TYPE_NAME)
+                    if tag_id:
+                        tag_name = self.client.atlan_tag_cache.get_name_for_id(tag_id)
+                        classification[self._TYPE_NAME] = (
+                            tag_name if tag_name else DELETED_
+                        )
+                        classification[self._TAG_ID] = tag_id
+                        # Check if the tag is a source tag (in that case tag has "attributes")
+                        attr_id = self.client.atlan_tag_cache.get_source_tags_attr_id(
+                            tag_id
+                        )
+                        if attr_id:
+                            classification[self._SOURCE_ATTACHMENTS] = [
+                                SourceTagAttachment(**source_tag["attributes"])
+                                for source_tag in classification.get("attributes").get(
+                                    attr_id
+                                )
+                                if isinstance(source_tag, dict)
+                                and source_tag.get("attributes")
+                            ]
 
         return raw_json