Merge pull request #117 from atlanhq/AM-448

Am 448

Author: ErnestoLoma

HISTORY.md

@@ -1,4 +1,10 @@
-## 0.8.0 (September 11, 20023)
+## 1.0.0 (September 13, 2023)
+* Initial production release
+* Adds column projections for listing users
+* Renamed register_client method to set_default_client from AtlanClient
+* Removed reset_default_client from AtlanClient
+
+## 0.8.0 (September 11, 2023)
 * Improve retry strategy for HTTP
 * Adds generators for developer portal docs
 * Adds create helper for Links

pyatlan/cache/atlan_tag_cache.py

@@ -3,7 +3,7 @@
 from typing import Optional
 
 from pyatlan.model.enums import AtlanTypeCategory
-from pyatlan.model.typedef import AtlanTagDef
+from pyatlan.model.typedef import AtlanTagDef, TypeDefResponseProvider
 
 
 class AtlanTagCache:
@@ -12,70 +12,105 @@ class AtlanTagCache:
     for Atlan tags.
     """
 
-    cache_by_id: dict[str, AtlanTagDef] = dict()
-    map_id_to_name: dict[str, str] = dict()
-    map_name_to_id: dict[str, str] = dict()
-    deleted_ids: set[str] = set()
-    deleted_names: set[str] = set()
+    caches: dict[int, "AtlanTagCache"] = {}
+
+    @classmethod
+    def get_cache(cls) -> "AtlanTagCache":
+        from pyatlan.client.atlan import AtlanClient
+
+        client = AtlanClient.get_default_client()
+        cache_key = client.cache_key
+        if cache_key not in cls.caches:
+            cls.caches[cache_key] = AtlanTagCache(provider=client)
+        return cls.caches[cache_key]
 
     @classmethod
     def refresh_cache(cls) -> None:
         """
         Refreshes the cache of Atlan tags by requesting the full set of Atlan tags from Atlan.
         """
-        from pyatlan.client.atlan import AtlanClient
+        cls.get_cache()._refresh_cache()
 
-        client = AtlanClient.get_default_client()
-        if client is None:
-            client = AtlanClient()
-        response = client.get_typedefs(type_category=AtlanTypeCategory.CLASSIFICATION)
+    @classmethod
+    def get_id_for_name(cls, name: str) -> Optional[str]:
+        """
+        Translate the provided human-readable Atlan tag name to its Atlan-internal ID string.
+
+        :param name: human-readable name of the Atlan tag
+        :returns: Atlan-internal ID string of the Atlan tag
+        """
+        return cls.get_cache()._get_id_for_name(name=name)
+
+    @classmethod
+    def get_name_for_id(cls, idstr: str) -> Optional[str]:
+        """
+        Translate the provided Atlan-internal classification ID string to the human-readable Atlan tag name.
+
+        :param idstr: Atlan-internal ID string of the Atlan tag
+        :returns: human-readable name of the Atlan tag
+        """
+        return cls.get_cache()._get_name_for_id(idstr=idstr)
+
+    def __init__(self, provider: TypeDefResponseProvider):
+        self.provider = provider
+        self.cache_by_id: dict[str, AtlanTagDef] = {}
+        self.map_id_to_name: dict[str, str] = {}
+        self.map_name_to_id: dict[str, str] = {}
+        self.deleted_ids: set[str] = set()
+        self.deleted_names: set[str] = set()
+
+    def _refresh_cache(self) -> None:
+        """
+        Refreshes the cache of Atlan tags by requesting the full set of Atlan tags from Atlan.
+        """
+        response = self.provider.get_typedefs(
+            type_category=AtlanTypeCategory.CLASSIFICATION
+        )
         if response is not None:
-            cls.cache_by_id = {}
-            cls.map_id_to_name = {}
-            cls.map_name_to_id = {}
+            self.cache_by_id = {}
+            self.map_id_to_name = {}
+            self.map_name_to_id = {}
             for atlan_tag in response.atlan_tag_defs:
                 atlan_tag_id = atlan_tag.name
                 atlan_tag_name = atlan_tag.display_name
-                cls.cache_by_id[atlan_tag_id] = atlan_tag
-                cls.map_id_to_name[atlan_tag_id] = atlan_tag_name
-                cls.map_name_to_id[atlan_tag_name] = atlan_tag_id
+                self.cache_by_id[atlan_tag_id] = atlan_tag
+                self.map_id_to_name[atlan_tag_id] = atlan_tag_name
+                self.map_name_to_id[atlan_tag_name] = atlan_tag_id
 
-    @classmethod
-    def get_id_for_name(cls, name: str) -> Optional[str]:
+    def _get_id_for_name(self, name: str) -> Optional[str]:
         """
         Translate the provided human-readable Atlan tag name to its Atlan-internal ID string.
 
         :param name: human-readable name of the Atlan tag
         :returns: Atlan-internal ID string of the Atlan tag
         """
-        cls_id = cls.map_name_to_id.get(name)
-        if not cls_id and name not in cls.deleted_names:
+        cls_id = self.map_name_to_id.get(name)
+        if not cls_id and name not in self.deleted_names:
             # If not found, refresh the cache and look again (could be stale)
-            cls.refresh_cache()
-            cls_id = cls.map_name_to_id.get(name)
+            self._refresh_cache()
+            cls_id = self.map_name_to_id.get(name)
             if not cls_id:
                 # If still not found after refresh, mark it as deleted (could be
                 # an entry in an audit log that refers to a classification that
                 # no longer exists)
-                cls.deleted_names.add(name)
+                self.deleted_names.add(name)
         return cls_id
 
-    @classmethod
-    def get_name_for_id(cls, idstr: str) -> Optional[str]:
+    def _get_name_for_id(self, idstr: str) -> Optional[str]:
         """
         Translate the provided Atlan-internal classification ID string to the human-readable Atlan tag name.
 
         :param idstr: Atlan-internal ID string of the Atlan tag
         :returns: human-readable name of the Atlan tag
         """
-        cls_name = cls.map_id_to_name.get(idstr)
-        if not cls_name and idstr not in cls.deleted_ids:
+        cls_name = self.map_id_to_name.get(idstr)
+        if not cls_name and idstr not in self.deleted_ids:
             # If not found, refresh the cache and look again (could be stale)
-            cls.refresh_cache()
-            cls_name = cls.map_id_to_name.get(idstr)
+            self._refresh_cache()
+            cls_name = self.map_id_to_name.get(idstr)
             if not cls_name:
                 # If still not found after refresh, mark it as deleted (could be
                 # an entry in an audit log that refers to a classification that
                 # no longer exists)
-                cls.deleted_ids.add(idstr)
+                self.deleted_ids.add(idstr)
         return cls_name