[feat] Added AtlanRequest which handles retranslators (human-readable to hashed names)

- Initial implementation of `AtlanRequest` to handle retranslators.
- Added `AtlanTagRetranslator` for tag name retranslation.

Author: Aryamanz29

pyatlan/client/atlan.py

@@ -81,7 +81,7 @@
     Purpose,
 )
 from pyatlan.model.atlan_image import AtlanImage
-from pyatlan.model.core import Announcement, AtlanObject, AtlanResponse
+from pyatlan.model.core import Announcement, AtlanObject, AtlanRequest, AtlanResponse
 from pyatlan.model.custom_metadata import CustomMetadataDict
 from pyatlan.model.enums import AtlanConnectorType, AtlanTypeCategory, CertificateStatus
 from pyatlan.model.group import AtlanGroup, CreateGroupResponse, GroupResponse
@@ -510,10 +510,13 @@ def _call_api_internal(
                     if text_response:
                         response_ = response.text
                     else:
-                        response_ = events if events else response.json()
-                        response_ = AtlanResponse(
-                            raw_json=response.json(), client=self
-                        ).to_dict()
+                        response_ = (
+                            events
+                            if events
+                            else AtlanResponse(
+                                raw_json=response.json(), client=self
+                            ).to_dict()
+                        )
                     LOGGER.debug("response: %s", response_)
                     return response_
                 except (
@@ -692,9 +695,7 @@ def _create_params(
             params["params"] = query_params
         if request_obj is not None:
             if isinstance(request_obj, AtlanObject):
-                params["data"] = request_obj.json(
-                    by_alias=True, exclude_unset=exclude_unset
-                )
+                params["data"] = AtlanRequest(request_obj, client=self).json()
             elif api.consumes == APPLICATION_ENCODED_FORM:
                 params["data"] = request_obj
             else:

pyatlan/model/core.py

@@ -24,6 +24,7 @@
 
 from pyatlan.model.constants import DELETED_, DELETED_SENTINEL
 from pyatlan.model.enums import AnnouncementType, EntityStatus, SaveSemantic
+from pyatlan.model.retranslators import AtlanTagRetranslator
 from pyatlan.model.structs import SourceTagAttachment
 from pyatlan.model.translators import AtlanTagTranslator
 
@@ -103,13 +104,13 @@ def _convert_to_tag_name(cls, data):
             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
+    # @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
-        )
+    #     return AtlanClient.get_current_client().atlan_tag_cache.get_id_for_name(
+    #         atlan_tag_name._display_text
+    #     )
 
 
 class AtlanObject(BaseModel):
@@ -219,6 +220,33 @@ def to_dict(self) -> Dict[str, Any]:
         return self.translated
 
 
+class AtlanRequest:
+    def __init__(self, model: BaseModel, client: "AtlanClient"):
+        self.client = client
+        self.model = model
+        self.retranslators = [
+            AtlanTagRetranslator(client),
+            # add others...
+        ]
+        # Do: model.json() → parse → translate → store
+        raw_json = self.model.json(by_alias=True, exclude_unset=True)
+        parsed = json.loads(raw_json)
+        self.translated = self._deep_retranslate(parsed)
+
+    def _deep_retranslate(self, data: Any) -> Any:
+        if isinstance(data, dict):
+            for retranslator in self.retranslators:
+                if retranslator.applies_to(data):
+                    data = retranslator.retranslate(data)
+            return {key: self._deep_retranslate(value) for key, value in data.items()}
+        elif isinstance(data, list):
+            return [self._deep_retranslate(item) for item in data]
+        return data
+
+    def json(self, **kwargs) -> str:
+        return json.dumps(self.translated, **kwargs)
+
+
 class SearchRequest(AtlanObject, ABC):
     attributes: Optional[List[str]] = Field(
         default_factory=list,

pyatlan/model/retranslators.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, Any, Dict
+
+if TYPE_CHECKING:
+    from pyatlan.client.atlan import AtlanClient
+
+
+class BaseRetranslator(ABC):
+    @abstractmethod
+    def applies_to(self, data: Dict[str, Any]) -> bool:
+        pass
+
+    @abstractmethod
+    def retranslate(self, data: Dict[str, Any]) -> Dict[str, Any]:
+        pass
+
+
+class AtlanTagRetranslator(BaseRetranslator):
+    _TYPE_NAME = "typeName"
+    _CLASSIFICATIONS = "classifications"
+    _CLASSIFICATION_NAMES = "classificationNames"
+    _SOURCE_ATTACHMENTS = "source_tag_attachements"
+
+    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
+
+    def retranslate(self, data: Dict[str, Any]) -> Dict[str, Any]:
+        data = data.copy()
+
+        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]
+            ]
+
+        if self._CLASSIFICATIONS in data:
+            for classification in data[self._CLASSIFICATIONS]:
+                tag_name = str(classification.get(self._TYPE_NAME))
+                if tag_name:
+                    tag_id = self.client.atlan_tag_cache.get_id_for_name(tag_name)
+                    classification[self._TYPE_NAME] = tag_id
+
+                    # Rebuild source tag structure if `source_tag_attachements` are present
+                    attachments = classification.pop(self._SOURCE_ATTACHMENTS, None)
+                    if attachments:
+                        attr_id = self.client.atlan_tag_cache.get_source_tags_attr_id(
+                            str(tag_id)
+                        )
+                        if attr_id:
+                            classification.setdefault("attributes", {})[attr_id] = [
+                                {
+                                    "typeName": "SourceTagAttachment",
+                                    "attributes": attachment.dict(),
+                                }
+                                for attachment in attachments
+                            ]
+
+        return data

pyatlan/model/utils.py

@@ -21,7 +21,7 @@ def encoders():
 
     return {
         datetime: lambda v: int(v.timestamp() * 1000),
-        AtlanTagName: AtlanTagName.json_encode_atlan_tag,
+        AtlanTagName: lambda atn: atn._display_text,
     }
 
 

tests/unit/test_atlan_tag_name.py

@@ -91,8 +91,8 @@ def test_convert_to_tag_name(current_client: AtlanClient, monkeypatch):
     assert str(sut) == ATLAN_TAG_ID
 
 
-def test_json_encode_atlan_tag(good_atlan_tag):
-    assert AtlanTagName.json_encode_atlan_tag(good_atlan_tag) == ATLAN_TAG_ID
+# def test_json_encode_atlan_tag(good_atlan_tag):
+#     assert AtlanTagName.json_encode_atlan_tag(good_atlan_tag) == ATLAN_TAG_ID
 
 
 def test_asset_tag_name_field_deserialization(current_client: AtlanClient, monkeypatch):