Skip to content

Cache

pyatlan.cache.atlan_tag_cache

Classes

AtlanTagCache(client: AtlanClient)

Lazily-loaded cache for translating between Atlan-internal ID strings and human-readable names for Atlan tags.

Source code in pyatlan/cache/atlan_tag_cache.py 
25
26
27
28
29
30
31
32
33
def __init__(self, client: AtlanClient):
    self.client: AtlanClient = client
    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()
    self.map_id_to_source_tags_attr_id: Dict[str, str] = {}
    self.lock: Lock = Lock()
Functions
get_id_for_name(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

Source code in pyatlan/cache/atlan_tag_cache.py 
41
42
43
44
45
46
47
48
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
    """
    return self._get_id_for_name(name=name)
get_name_for_id(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

Source code in pyatlan/cache/atlan_tag_cache.py 
50
51
52
53
54
55
56
57
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
    """
    return self._get_name_for_id(idstr=idstr)
get_source_tags_attr_id(id: str) -> Optional[str]

Translate the provided Atlan-internal Atlan tag ID string to the Atlan-internal name of the attribute that captures tag attachment details (for source-synced tags).

:param id: Atlan-internal ID string of the Atlan tag :returns: Atlan-internal ID string of the attribute containing source-synced tag attachment details

Source code in pyatlan/cache/atlan_tag_cache.py 
59
60
61
62
63
64
65
66
67
def get_source_tags_attr_id(self, id: str) -> Optional[str]:
    """
    Translate the provided Atlan-internal Atlan tag ID string to the Atlan-internal name of the attribute that
    captures tag attachment details (for source-synced tags).

    :param id: Atlan-internal ID string of the Atlan tag
    :returns: Atlan-internal ID string of the attribute containing source-synced tag attachment details
    """
    return self._get_source_tags_attr_id(id)
refresh_cache() -> None

Refreshes the cache of Atlan tags by requesting the full set of Atlan tags from Atlan.

Source code in pyatlan/cache/atlan_tag_cache.py 
35
36
37
38
39
def refresh_cache(self) -> None:
    """
    Refreshes the cache of Atlan tags by requesting the full set of Atlan tags from Atlan.
    """
    self._refresh_cache()

pyatlan.cache.custom_metadata_cache

Classes

CustomMetadataCache(client: AtlanClient)

Lazily-loaded cache for translating between Atlan-internal ID strings and human-readable names for custom metadata (including attributes).

Source code in pyatlan/cache/custom_metadata_cache.py 
24
25
26
27
28
29
30
31
32
33
34
def __init__(self, client: AtlanClient):
    self.client: AtlanClient = client
    self.cache_by_id: Dict[str, CustomMetadataDef] = {}
    self.attr_cache_by_id: Dict[str, AttributeDef] = {}
    self.map_id_to_name: Dict[str, str] = {}
    self.map_name_to_id: Dict[str, str] = {}
    self.map_attr_id_to_name: Dict[str, Dict[str, str]] = {}
    self.map_attr_name_to_id: Dict[str, Dict[str, str]] = {}
    self.archived_attr_ids: Dict[str, str] = {}
    self.types_by_asset: Dict[str, Set[type]] = {}
    self.lock: Lock = Lock()
Functions
get_all_custom_attributes(include_deleted: bool = False, force_refresh: bool = False) -> Dict[str, List[AttributeDef]]

Retrieve all the custom metadata attributes. The dict will be keyed by custom metadata set name, and the value will be a listing of all the attributes within that set (with all the details of each of those attributes).

:param include_deleted: if True, include the archived (deleted) custom attributes; otherwise only include active custom attributes :param force_refresh: if True, will refresh the custom metadata cache; if False, will only refresh the cache if it is empty :returns: a dict from custom metadata set name to all details about its attributes :raises NotFoundError: if the custom metadata cannot be found

Source code in pyatlan/cache/custom_metadata_cache.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def get_all_custom_attributes(
    self, include_deleted: bool = False, force_refresh: bool = False
) -> Dict[str, List[AttributeDef]]:
    """
    Retrieve all the custom metadata attributes. The dict will be keyed by custom metadata set
    name, and the value will be a listing of all the attributes within that set (with all the details
    of each of those attributes).

    :param include_deleted: if True, include the archived (deleted) custom attributes; otherwise only
                            include active custom attributes
    :param force_refresh: if True, will refresh the custom metadata cache; if False, will only refresh the
                          cache if it is empty
    :returns: a dict from custom metadata set name to all details about its attributes
    :raises NotFoundError: if the custom metadata cannot be found
    """
    return self._get_all_custom_attributes(
        include_deleted=include_deleted, force_refresh=force_refresh
    )
get_attr_id_for_name(set_name: str, attr_name: str) -> str

Translate the provided human-readable custom metadata set and attribute names to the Atlan-internal ID string for the attribute.

:param set_name: human-readable name of the custom metadata set :param attr_name: human-readable name of the attribute :returns: Atlan-internal ID string for the attribute :raises NotFoundError: if the custom metadata attribute cannot be found

Source code in pyatlan/cache/custom_metadata_cache.py 
85
86
87
88
89
90
91
92
93
94
95
def get_attr_id_for_name(self, set_name: str, attr_name: str) -> str:
    """
    Translate the provided human-readable custom metadata set and attribute names to the Atlan-internal ID string
    for the attribute.

    :param set_name: human-readable name of the custom metadata set
    :param attr_name: human-readable name of the attribute
    :returns: Atlan-internal ID string for the attribute
    :raises NotFoundError: if the custom metadata attribute cannot be found
    """
    return self._get_attr_id_for_name(set_name=set_name, attr_name=attr_name)
get_attr_name_for_id(set_id: str, attr_id: str) -> str

Given the Atlan-internal ID string for the set and the Atlan-internal ID for the attribute return the human-readable custom metadata name for the attribute.

:param set_id: Atlan-internal ID string for the custom metadata set :param attr_id: Atlan-internal ID string for the attribute :returns: human-readable name of the attribute :raises NotFoundError: if the custom metadata attribute cannot be found

Source code in pyatlan/cache/custom_metadata_cache.py 
 97
 98
 99
100
101
102
103
104
105
106
107
def get_attr_name_for_id(self, set_id: str, attr_id: str) -> str:
    """
    Given the Atlan-internal ID string for the set and the Atlan-internal ID for the attribute return the
    human-readable custom metadata name for the attribute.

    :param set_id: Atlan-internal ID string for the custom metadata set
    :param attr_id: Atlan-internal ID string for the attribute
    :returns: human-readable name of the attribute
    :raises NotFoundError: if the custom metadata attribute cannot be found
    """
    return self._get_attr_name_for_id(set_id=set_id, attr_id=attr_id)
get_attribute_def(attr_id: str) -> AttributeDef

Retrieve a specific custom metadata attribute definition by its unique Atlan-internal ID string.

:param attr_id: Atlan-internal ID string for the custom metadata attribute :returns: attribute definition for the custom metadata attribute :raises InvalidRequestError: if no attribute ID was provided :raises NotFoundError: if the custom metadata attribute cannot be found

Source code in pyatlan/cache/custom_metadata_cache.py 
152
153
154
155
156
157
158
159
160
161
def get_attribute_def(self, attr_id: str) -> AttributeDef:
    """
    Retrieve a specific custom metadata attribute definition by its unique Atlan-internal ID string.

    :param attr_id: Atlan-internal ID string for the custom metadata attribute
    :returns: attribute definition for the custom metadata attribute
    :raises InvalidRequestError: if no attribute ID was provided
    :raises NotFoundError: if the custom metadata attribute cannot be found
    """
    return self._get_attribute_def(attr_id=attr_id)
get_attribute_for_search_results(set_name: str, attr_name: str) -> Optional[str]

Retrieve a single custom attribute name to include on search results.

:param set_name: human-readable name of the custom metadata set for which to retrieve the custom metadata attribute name :param attr_name: human-readable name of the attribute :returns: the attribute name, strictly useful for inclusion in search results

Source code in pyatlan/cache/custom_metadata_cache.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def get_attribute_for_search_results(
    self, set_name: str, attr_name: str
) -> Optional[str]:
    """
    Retrieve a single custom attribute name to include on search results.

    :param set_name: human-readable name of the custom metadata set for which to retrieve the custom metadata
                     attribute name
    :param attr_name: human-readable name of the attribute
    :returns: the attribute name, strictly useful for inclusion in search results
    """
    return self._get_attribute_for_search_results(
        set_name=set_name, attr_name=attr_name
    )
get_attributes_for_search_results(set_name: str) -> Optional[List[str]]

Retrieve the full set of custom attributes to include on search results.

:param set_name: human-readable name of the custom metadata set for which to retrieve attribute names :returns: a list of the attribute names, strictly useful for inclusion in search results

Source code in pyatlan/cache/custom_metadata_cache.py 
117
118
119
120
121
122
123
124
def get_attributes_for_search_results(self, set_name: str) -> Optional[List[str]]:
    """
    Retrieve the full set of custom attributes to include on search results.

    :param set_name: human-readable name of the custom metadata set for which to retrieve attribute names
    :returns: a list of the attribute names, strictly useful for inclusion in search results
    """
    return self._get_attributes_for_search_results(set_name=set_name)
get_custom_metadata_def(name: str) -> CustomMetadataDef

Retrieve the full custom metadata structure definition.

:param name: human-readable name of the custom metadata set :returns: the full custom metadata structure definition for that set :raises InvalidRequestError: if no name was provided :raises NotFoundError: if the custom metadata cannot be found

Source code in pyatlan/cache/custom_metadata_cache.py 
141
142
143
144
145
146
147
148
149
150
def get_custom_metadata_def(self, name: str) -> CustomMetadataDef:
    """
    Retrieve the full custom metadata structure definition.

    :param name: human-readable name of the custom metadata set
    :returns: the full custom metadata structure definition for that set
    :raises InvalidRequestError: if no name was provided
    :raises NotFoundError: if the custom metadata cannot be found
    """
    return self._get_custom_metadata_def(name=name)
get_id_for_name(name: str) -> str

Translate the provided human-readable custom metadata set name to its Atlan-internal ID string.

:param name: human-readable name of the custom metadata set :returns: Atlan-internal ID string of the custom metadata set :raises InvalidRequestError: if no name was provided :raises NotFoundError: if the custom metadata cannot be found

Source code in pyatlan/cache/custom_metadata_cache.py 
44
45
46
47
48
49
50
51
52
53
def get_id_for_name(self, name: str) -> str:
    """
    Translate the provided human-readable custom metadata set name to its Atlan-internal ID string.

    :param name: human-readable name of the custom metadata set
    :returns: Atlan-internal ID string of the custom metadata set
    :raises InvalidRequestError: if no name was provided
    :raises NotFoundError: if the custom metadata cannot be found
    """
    return self._get_id_for_name(name=name)
get_name_for_id(idstr: str) -> str

Translate the provided Atlan-internal custom metadata ID string to the human-readable custom metadata set name.

:param idstr: Atlan-internal ID string of the custom metadata set :returns: human-readable name of the custom metadata set :raises InvalidRequestError: if no ID was provided :raises NotFoundError: if the custom metadata cannot be found

Source code in pyatlan/cache/custom_metadata_cache.py 
55
56
57
58
59
60
61
62
63
64
def get_name_for_id(self, idstr: str) -> str:
    """
    Translate the provided Atlan-internal custom metadata ID string to the human-readable custom metadata set name.

    :param idstr: Atlan-internal ID string of the custom metadata set
    :returns: human-readable name of the custom metadata set
    :raises InvalidRequestError: if no ID was provided
    :raises NotFoundError: if the custom metadata cannot be found
    """
    return self._get_name_for_id(idstr=idstr)
is_attr_archived(attr_id: str) -> bool

Determine if an attribute is archived :param attr_id: Atlan-internal ID string for the attribute :returns: True if the attribute has been archived

Source code in pyatlan/cache/custom_metadata_cache.py 
109
110
111
112
113
114
115
def is_attr_archived(self, attr_id: str) -> bool:
    """
    Determine if an attribute is archived
    :param attr_id: Atlan-internal ID string for the attribute
    :returns: True if the attribute has been archived
    """
    return self._is_attr_archived(attr_id=attr_id)
refresh_cache() -> None

Refreshes the cache of custom metadata structures by requesting the full set of custom metadata structures from Atlan. :raises LogicError: if duplicate custom attributes are detected

Source code in pyatlan/cache/custom_metadata_cache.py 
36
37
38
39
40
41
42
def refresh_cache(self) -> None:
    """
    Refreshes the cache of custom metadata structures by requesting the full set of custom metadata
    structures from Atlan.
    :raises LogicError: if duplicate custom attributes are detected
    """
    self._refresh_cache()

pyatlan.cache.enum_cache

Classes

EnumCache(client: AtlanClient)

Lazily-loaded cache for accessing details of an enumeration.

Source code in pyatlan/cache/enum_cache.py 
24
25
26
27
def __init__(self, client: AtlanClient):
    self.client: AtlanClient = client
    self.cache_by_name: Dict[str, EnumDef] = {}
    self.lock: Lock = Lock()
Functions
get_by_name(name: str) -> EnumDef

Retrieve the enumeration definition by its name.

:param name: human-readable name of the enumeration. :raises NotFoundError: if the enumeration with the given name does not exist. :returns: enumeration definition

Source code in pyatlan/cache/enum_cache.py 
29
30
31
32
33
34
35
36
37
38
39
def get_by_name(self, name: str) -> EnumDef:
    """
    Retrieve the enumeration definition by its name.

    :param name: human-readable name of the enumeration.
    :raises `NotFoundError`: if the enumeration with the given name does not exist.
    :returns: enumeration definition
    """
    if not (enum := self._get_by_name(name=name)):
        raise ErrorCode.ENUM_NOT_FOUND.exception_with_parameters(name)
    return enum
refresh_cache() -> None

Refreshes the cache of enumerations by requesting the full set of enumerations from Atlan.

Source code in pyatlan/cache/enum_cache.py 
41
42
43
44
45
46
47
48
49
50
51
52
def refresh_cache(self) -> None:
    """
    Refreshes the cache of enumerations by requesting the full set of enumerations from Atlan.
    """
    with self.lock:
        # Make API call directly
        response = self.client.typedef.get(type_category=AtlanTypeCategory.ENUM)
        if not response or not response.enum_defs:
            raise ErrorCode.EXPIRED_API_TOKEN.exception_with_parameters()

        # Process response using shared logic
        self.cache_by_name = EnumCacheCommon.refresh_cache_data(response)

pyatlan.cache.group_cache

Classes

GroupCache(client: AtlanClient)

Lazily-loaded cache for translating Atlan-internal groups into their various IDs.

Source code in pyatlan/cache/group_cache.py 
23
24
25
26
27
28
def __init__(self, client: AtlanClient):
    self.client: AtlanClient = client
    self.map_id_to_name: Dict[str, str] = {}
    self.map_name_to_id: Dict[str, str] = {}
    self.map_alias_to_id: Dict[str, str] = {}
    self.lock: Lock = Lock()
Functions
get_id_for_alias(alias: str) -> Optional[str]

Translate the provided human-readable group name to its GUID.

:param alias: name of the group as it appears in the UI :returns: unique identifier (GUID) of the group

Source code in pyatlan/cache/group_cache.py 
39
40
41
42
43
44
45
46
def get_id_for_alias(self, alias: str) -> Optional[str]:
    """
    Translate the provided human-readable group name to its GUID.

    :param alias: name of the group as it appears in the UI
    :returns: unique identifier (GUID) of the group
    """
    return self._get_id_for_alias(alias=alias)
get_id_for_name(name: str) -> Optional[str]

Translate the provided internal group name to its GUID.

:param name: human-readable name of the group :returns: unique identifier (GUID) of the group

Source code in pyatlan/cache/group_cache.py 
30
31
32
33
34
35
36
37
def get_id_for_name(self, name: str) -> Optional[str]:
    """
    Translate the provided internal group name to its GUID.

    :param name: human-readable name of the group
    :returns: unique identifier (GUID) of the group
    """
    return self._get_id_for_name(name=name)
get_name_for_id(idstr: str) -> Optional[str]

Translate the provided group GUID to the internal group name.

:param idstr: unique identifier (GUID) of the group :returns: human-readable name of the group

Source code in pyatlan/cache/group_cache.py 
48
49
50
51
52
53
54
55
def get_name_for_id(self, idstr: str) -> Optional[str]:
    """
    Translate the provided group GUID to the internal group name.

    :param idstr: unique identifier (GUID) of the group
    :returns: human-readable name of the group
    """
    return self._get_name_for_id(idstr=idstr)
validate_aliases(aliases: Iterable[str])

Validate that the given (internal) group names are valid. A ValueError will be raised in any are not.

:param aliases: a collection of (internal) group names to be checked

Source code in pyatlan/cache/group_cache.py 
57
58
59
60
61
62
63
def validate_aliases(self, aliases: Iterable[str]):
    """
    Validate that the given (internal) group names are valid. A ValueError will be raised in any are not.

    :param aliases: a collection of (internal) group names to be checked
    """
    return self._validate_aliases(aliases)

pyatlan.cache.role_cache

Classes

RoleCache(client: AtlanClient)

Lazily-loaded cache for translating Atlan-internal roles into their various IDs.

Source code in pyatlan/cache/role_cache.py 
23
24
25
26
27
28
29
def __init__(self, client: AtlanClient):
    self.client: AtlanClient = client
    self.cache_by_id: Dict[str, AtlanRole] = {}
    self.map_id_to_name: Dict[str, str] = {}
    self.map_name_to_id: Dict[str, str] = {}
    self.lock: Lock = Lock()
    self._is_api_token_user: Optional[bool] = None
Functions
get_id_for_name(name: str) -> Optional[str]

Translate the provided human-readable role name to its GUID.

:param name: human-readable name of the role :returns: unique identifier (GUID) of the role

Source code in pyatlan/cache/role_cache.py 
31
32
33
34
35
36
37
38
def get_id_for_name(self, name: str) -> Optional[str]:
    """
    Translate the provided human-readable role name to its GUID.

    :param name: human-readable name of the role
    :returns: unique identifier (GUID) of the role
    """
    return self._get_id_for_name(name=name)
get_name_for_id(idstr: str) -> Optional[str]

Translate the provided role GUID to the human-readable role name.

:param idstr: unique identifier (GUID) of the role :returns: human-readable name of the role

Source code in pyatlan/cache/role_cache.py 
40
41
42
43
44
45
46
47
def get_name_for_id(self, idstr: str) -> Optional[str]:
    """
    Translate the provided role GUID to the human-readable role name.

    :param idstr: unique identifier (GUID) of the role
    :returns: human-readable name of the role
    """
    return self._get_name_for_id(idstr=idstr)
is_api_token_user() -> bool

Check if the current user is authenticated via an API token or OAuth client. This method checks for the presence of $api-token-default-access or $admin roles.

:returns: True if the user is an API token user, False otherwise

Source code in pyatlan/cache/role_cache.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def is_api_token_user(self) -> bool:
    """
    Check if the current user is authenticated via an API token or OAuth client.
    This method checks for the presence of $api-token-default-access or $admin roles.

    :returns: True if the user is an API token user, False otherwise
    """
    if self._is_api_token_user is not None:
        return self._is_api_token_user

    # Fetch role mappings for the current user
    raw_json = self.client._call_api(api=GET_WHOAMI_USER)

    # Check if the user has $api-token-default-access or $admin roles
    for role_mapping in raw_json["roles"]:
        role_name = role_mapping.get("name")
        if role_name in ["$api-token-default-access", "$admin"]:
            self._is_api_token_user = True
            return self._is_api_token_user

    self._is_api_token_user = False
    return self._is_api_token_user
validate_idstrs(idstrs: Iterable[str])

Validate that the given role GUIDs are valid. A ValueError will be raised in any are not.

:param idstrs: a collection of unique identifiers (GUID) of the roles to be checked

Source code in pyatlan/cache/role_cache.py 
49
50
51
52
53
54
55
def validate_idstrs(self, idstrs: Iterable[str]):
    """
    Validate that the given role GUIDs are valid. A ValueError will be raised in any are not.

    :param idstrs: a collection of unique identifiers (GUID) of the roles to be checked
    """
    return self._validate_idstrs(idstrs=idstrs)

pyatlan.cache.user_cache

Classes

UserCache(client: AtlanClient)

Lazily-loaded cache for translating Atlan-internal users into their various IDs.

Source code in pyatlan/cache/user_cache.py 
23
24
25
26
27
28
def __init__(self, client: AtlanClient):
    self.client: AtlanClient = client
    self.map_id_to_name: Dict[str, str] = {}
    self.map_name_to_id: Dict[str, str] = {}
    self.map_email_to_id: Dict[str, str] = {}
    self.lock: Lock = Lock()
Functions
get_id_for_email(email: str) -> Optional[str]

Translate the provided email to its GUID.

:param email: email address of the user :returns: unique identifier (GUID) of the user

Source code in pyatlan/cache/user_cache.py 
39
40
41
42
43
44
45
46
def get_id_for_email(self, email: str) -> Optional[str]:
    """
    Translate the provided email to its GUID.

    :param email: email address of the user
    :returns: unique identifier (GUID) of the user
    """
    return self._get_id_for_email(email=email)
get_id_for_name(name: str) -> Optional[str]

Translate the provided human-readable username to its GUID.

:param name: human-readable name of the user :returns: unique identifier (GUID) of the user

Source code in pyatlan/cache/user_cache.py 
30
31
32
33
34
35
36
37
def get_id_for_name(self, name: str) -> Optional[str]:
    """
    Translate the provided human-readable username to its GUID.

    :param name: human-readable name of the user
    :returns: unique identifier (GUID) of the user
    """
    return self._get_id_for_name(name=name)
get_name_for_id(idstr: str) -> Optional[str]

Translate the provided user GUID to the human-readable username.

:param idstr: unique identifier (GUID) of the user :returns: username of the user

Source code in pyatlan/cache/user_cache.py 
48
49
50
51
52
53
54
55
def get_name_for_id(self, idstr: str) -> Optional[str]:
    """
    Translate the provided user GUID to the human-readable username.

    :param idstr: unique identifier (GUID) of the user
    :returns: username of the user
    """
    return self._get_name_for_id(idstr=idstr)
validate_names(names: Iterable[str])

Validate that the given human-readable usernames are valid. A ValueError will be raised in any are not.

:param names: a collection of usernames to be checked

Source code in pyatlan/cache/user_cache.py 
57
58
59
60
61
62
63
def validate_names(self, names: Iterable[str]):
    """
    Validate that the given human-readable usernames are valid. A ValueError will be raised in any are not.

    :param names: a collection of usernames to be checked
    """
    return self._validate_names(names)

pyatlan.cache.connection_cache

Classes

ConnectionCache(client: AtlanClient)

Bases: AbstractAssetCache

Lazily-loaded cache for translating between a connection's simplified name its details.

  • guid = UUID of the connection for eg: 9c677e77-e01d-40e0-85b7-8ba4cd7d0ea9
  • qualified_name = Atlan-internal name of the connection (with epoch) for eg: default/snowflake/1234567890
  • name = simple name of the form {{connectorType}}/{{connectorName}}, for eg: snowflake/development
Source code in pyatlan/cache/connection_cache.py 
42
43
def __init__(self, client: AtlanClient):
    super().__init__(client)
Functions
get_by_guid(guid: str, allow_refresh: bool = True) -> Connection

Retrieve a connection from the cache by its UUID. If the asset is not found, it will be looked up and added to the cache.

:param guid: UUID of the connection in Atlan for eg: 9c677e77-e01d-40e0-85b7-8ba4cd7d0ea9 :returns: connection (if found) :raises AtlanError: on any API communication problem if the cache needs to be refreshed :raises NotFoundError: if the connection cannot be found (does not exist) in Atlan :raises InvalidRequestError: if no UUID was provided for the connection to retrieve

Source code in pyatlan/cache/connection_cache.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
def get_by_guid(self, guid: str, allow_refresh: bool = True) -> Connection:
    """
    Retrieve a connection from the cache by its UUID.
    If the asset is not found, it will be looked up and added to the cache.

    :param guid: UUID of the connection in Atlan
        for eg: 9c677e77-e01d-40e0-85b7-8ba4cd7d0ea9
    :returns: connection (if found)
    :raises AtlanError: on any API communication problem if the cache needs to be refreshed
    :raises NotFoundError: if the connection cannot be found (does not exist) in Atlan
    :raises InvalidRequestError: if no UUID was provided for the connection to retrieve
    """
    return self._get_by_guid(guid=guid, allow_refresh=allow_refresh)
get_by_name(name: ConnectionName, allow_refresh: bool = True) -> Connection

Retrieve an connection from the cache by its uniquely identifiable name.

:param name: uniquely identifiable name of the connection in Atlan In the form of {{connectorType}}/{{connectorName}} for eg: snowflake/development :param allow_refresh: whether to allow a refresh of the cache (True) or not (False) :returns: connection (if found) :raises AtlanError: on any API communication problem if the cache needs to be refreshed :raises NotFoundError: if the connection cannot be found (does not exist) in Atlan :raises InvalidRequestError: if no name was provided for the connection to retrieve

Source code in pyatlan/cache/connection_cache.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def get_by_name(
    self, name: ConnectionName, allow_refresh: bool = True
) -> Connection:
    """
    Retrieve an connection from the cache by its uniquely identifiable name.

    :param name: uniquely identifiable name of the connection in Atlan
        In the form of {{connectorType}}/{{connectorName}}
        for eg: snowflake/development
    :param allow_refresh: whether to allow a refresh of the cache (`True`) or not (`False`)
    :returns: connection (if found)
    :raises AtlanError: on any API communication problem if the cache needs to be refreshed
    :raises NotFoundError: if the connection cannot be found (does not exist) in Atlan
    :raises InvalidRequestError: if no name was provided for the connection to retrieve
    """
    return self._get_by_name(name=name, allow_refresh=allow_refresh)
get_by_qualified_name(qualified_name: str, allow_refresh: bool = True) -> Connection

Retrieve a connection from the cache by its unique Atlan-internal name.

:param qualified_name: unique Atlan-internal name of the connection for eg: default/snowflake/1234567890 :param allow_refresh: whether to allow a refresh of the cache (True) or not (False) :param qualified_name: unique Atlan-internal name of the connection :returns: connection (if found) :raises AtlanError: on any API communication problem if the cache needs to be refreshed :raises NotFoundError: if the connection cannot be found (does not exist) in Atlan :raises InvalidRequestError: if no qualified_name was provided for the connection to retrieve

Source code in pyatlan/cache/connection_cache.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def get_by_qualified_name(
    self, qualified_name: str, allow_refresh: bool = True
) -> Connection:
    """
    Retrieve a connection from the cache by its unique Atlan-internal name.

    :param qualified_name: unique Atlan-internal name of the connection
        for eg: default/snowflake/1234567890
    :param allow_refresh: whether to allow a refresh of the cache (`True`) or not (`False`)
    :param qualified_name: unique Atlan-internal name of the connection
    :returns: connection (if found)
    :raises AtlanError: on any API communication problem if the cache needs to be refreshed
    :raises NotFoundError: if the connection cannot be found (does not exist) in Atlan
    :raises InvalidRequestError: if no qualified_name was provided for the connection to retrieve
    """
    return self._get_by_qualified_name(
        qualified_name=qualified_name, allow_refresh=allow_refresh
    )

ConnectionName(connection: Union[str, Optional[Connection]] = None)

Bases: AbstractAssetName

Unique identity for a connection, in the form: {{type}}/{{name}}

  • For eg: snowflake/development
Source code in pyatlan/cache/connection_cache.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def __init__(
    self,
    connection: Union[
        str,
        Optional[Connection],
    ] = None,
):
    self.name = None
    self.type = None

    if isinstance(connection, Connection):
        self.name = connection.name
        self.type = connection.connector_name

    elif isinstance(connection, str):
        tokens = connection.split("/")
        if len(tokens) > 1:
            # Try enum conversion; fallback to custom connector if it fails
            try:
                self.type = AtlanConnectorType(tokens[0]).value  # type: ignore[call-arg]
                self.name = connection[len(tokens[0]) + 1 :]  # noqa
            except ValueError:
                custom_connector = AtlanConnectorType.CREATE_CUSTOM(
                    # Ensure the enum name is converted to UPPER_SNAKE_CASE from kebab-case
                    name=tokens[0].replace("-", "_").upper(),
                    value=tokens[0],
                )
                self.type = custom_connector.value
                self.name = connection[len(tokens[0]) + 1 :]

pyatlan.cache.source_tag_cache

Classes

SourceTagCache(client: AtlanClient)

Bases: AbstractAssetCache

Lazily-loaded cache for translating between source-synced tags and the qualifiedName of such.

  • guid = UUID of the source tag for eg: 9c677e77-e01d-40e0-85b7-8ba4cd7d0ea9
  • qualified_name = of the source tag (with epoch) for eg: default/snowflake/1234567890/DB/SCHEMA/TAG_NAME
  • name = simple name of the form {{connectorType}}/{{connectorName}}@@DB/SCHEMA/TAG_NAME for eg: snowflake/development@@DB/SCHEMA/TAG_NAME
Source code in pyatlan/cache/source_tag_cache.py 
39
40
def __init__(self, client: AtlanClient):
    super().__init__(client)
Functions
get_by_guid(guid: str, allow_refresh: bool = True) -> Tag

Retrieve a source tag from the cache by its UUID. If the asset is not found, it will be looked up and added to the cache.

:param guid: UUID of the source tag in Atlan for eg: 9c677e77-e01d-40e0-85b7-8ba4cd7d0ea9 :returns: source tag (if found) :raises AtlanError: on any API communication problem if the cache needs to be refreshed :raises NotFoundError: if the source tag cannot be found (does not exist) in Atlan :raises InvalidRequestError: if no UUID was provided for the source tag to retrieve

Source code in pyatlan/cache/source_tag_cache.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
def get_by_guid(self, guid: str, allow_refresh: bool = True) -> Tag:
    """
    Retrieve a source tag from the cache by its UUID.
    If the asset is not found, it will be looked up and added to the cache.

    :param guid: UUID of the source tag in Atlan
        for eg: 9c677e77-e01d-40e0-85b7-8ba4cd7d0ea9
    :returns: source tag (if found)
    :raises AtlanError: on any API communication problem if the cache needs to be refreshed
    :raises NotFoundError: if the source tag cannot be found (does not exist) in Atlan
    :raises InvalidRequestError: if no UUID was provided for the source tag to retrieve
    """
    return self._get_by_guid(guid=guid, allow_refresh=allow_refresh)
get_by_name(name: SourceTagName, allow_refresh: bool = True) -> Tag

Retrieve an connection from the cache by its uniquely identifiable name.

:param name: uniquely identifiable name of the connection in Atlan. In the form of {{connectorType}}/{{connectorName}}@@DB/SCHEMA/TAG_NAME for eg: snowflake/development@@DB/SCHEMA/TAG_NAME :param allow_refresh: whether to allow a refresh of the cache (True) or not (False) :returns: the connection (if found) :raises AtlanError: on any API communication problem if the cache needs to be refreshed :raises NotFoundError: if the object cannot be found (does not exist) in Atlan :raises InvalidRequestError: if no name was provided for the object to retrieve

Source code in pyatlan/cache/source_tag_cache.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def get_by_name(self, name: SourceTagName, allow_refresh: bool = True) -> Tag:
    """
    Retrieve an connection from the cache by its uniquely identifiable name.

    :param name: uniquely identifiable name of the connection in Atlan.
        In the form of {{connectorType}}/{{connectorName}}@@DB/SCHEMA/TAG_NAME
        for eg: snowflake/development@@DB/SCHEMA/TAG_NAME
    :param allow_refresh: whether to allow a refresh of the cache (`True`) or not (`False`)
    :returns: the connection (if found)
    :raises AtlanError: on any API communication problem if the cache needs to be refreshed
    :raises NotFoundError: if the object cannot be found (does not exist) in Atlan
    :raises InvalidRequestError: if no name was provided for the object to retrieve
    """
    return self._get_by_name(name=name, allow_refresh=allow_refresh)
get_by_qualified_name(qualified_name: str, allow_refresh: bool = True) -> Tag

Retrieve a source tag from the cache by its unique Atlan-internal name.

:param qualified_name: unique Atlan-internal name of the source tag for eg: default/snowflake/1234567890/DB/SCHEMA/TAG_NAME :param allow_refresh: whether to allow a refresh of the cache (True) or not (False) :param qualified_name: unique Atlan-internal name of the source tag :returns: source tag (if found) :raises AtlanError: on any API communication problem if the cache needs to be refreshed :raises NotFoundError: if the source tag cannot be found (does not exist) in Atlan :raises InvalidRequestError: if no qualified_name was provided for the source tag to retrieve

Source code in pyatlan/cache/source_tag_cache.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def get_by_qualified_name(
    self, qualified_name: str, allow_refresh: bool = True
) -> Tag:
    """
    Retrieve a source tag from the cache by its unique Atlan-internal name.

    :param qualified_name: unique Atlan-internal name of the source tag
        for eg: default/snowflake/1234567890/DB/SCHEMA/TAG_NAME
    :param allow_refresh: whether to allow a refresh of the cache (`True`) or not (`False`)
    :param qualified_name: unique Atlan-internal name of the source tag
    :returns: source tag (if found)
    :raises AtlanError: on any API communication problem if the cache needs to be refreshed
    :raises NotFoundError: if the source tag cannot be found (does not exist) in Atlan
    :raises InvalidRequestError: if no qualified_name was provided for the source tag to retrieve
    """
    return self._get_by_qualified_name(
        qualified_name=qualified_name, allow_refresh=allow_refresh
    )

SourceTagName(client: AtlanClient, tag: Union[str, Asset])

Bases: AbstractAssetName

Unique identity for a source tag, in the form: {{connectorType}}/{{connectorName}}@@DB/SCHEMA/TAG_NAME

  • For eg: snowflake/development
Source code in pyatlan/cache/source_tag_cache.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def __init__(self, client: AtlanClient, tag: Union[str, Asset]):
    self.connection = None
    self.partial_tag_name = None

    # NOTE: Checking if the first result is an "Asset" since in pyatlan,
    # "DbtTag" extends "Dbt" (unlike other tags like "SnowflakeTag" that extend the "Tag" model),
    # preventing Dbt tags from being excluded from caching:
    if isinstance(tag, Asset):
        source_tag_qn = tag.qualified_name or ""
        tokens = source_tag_qn.split("/")
        connection_qn = "/".join(tokens[:3]) if len(tokens) >= 3 else ""
        conn = client.connection_cache.get_by_qualified_name(connection_qn)
        self.connection = ConnectionName(conn)
        self.partial_tag_name = source_tag_qn[len(connection_qn) + 1 :]  # noqa

    elif isinstance(tag, str):
        tokens = tag.split(self._CONNECTION_DELIMITER)
        if len(tokens) == 2:
            self.connection = ConnectionName(tokens[0])
            self.partial_tag_name = tokens[1]

pyatlan.cache.abstract_asset_cache

Classes

AbstractAssetCache(client: AtlanClient)

Bases: ABC

Base class for reusable components that are common to all caches, where a cache is populated entry-by-entry.

Source code in pyatlan/cache/abstract_asset_cache.py 
23
24
25
26
27
28
def __init__(self, client: AtlanClient):
    self.client = client
    self.lock = threading.Lock()
    self.name_to_guid: Dict[str, str] = dict()
    self.guid_to_asset: Dict[str, Asset] = dict()
    self.qualified_name_to_guid: Dict[str, str] = dict()
Functions
cache(asset: Asset)

Add an entry to the cache.

:param asset: to be cached

Source code in pyatlan/cache/abstract_asset_cache.py 
76
77
78
79
80
81
82
83
84
85
86
87
def cache(self, asset: Asset):
    """
    Add an entry to the cache.

    :param asset: to be cached
    """
    name = asset and self.get_name(asset)
    if not all([name, asset.guid, asset.qualified_name]):
        return
    self.name_to_guid[name] = asset.guid  # type: ignore[index]
    self.guid_to_asset[asset.guid] = asset  # type: ignore[index]
    self.qualified_name_to_guid[asset.qualified_name] = asset.guid  # type: ignore[index]
get_name(asset: Asset) abstractmethod

Abstract method to get name from asset.

Source code in pyatlan/cache/abstract_asset_cache.py 
42
43
44
@abstractmethod
def get_name(self, asset: Asset):
    """Abstract method to get name from asset."""
is_guid_known(guid: str) -> bool

Checks whether the provided Atlan-internal UUID is known. NOTE: will not refresh the cache itself to determine this.

:param guid: Atlan-internal UUID of the object :returns: True if the object is known, False otherwise

Source code in pyatlan/cache/abstract_asset_cache.py 
46
47
48
49
50
51
52
53
54
def is_guid_known(self, guid: str) -> bool:
    """
    Checks whether the provided Atlan-internal UUID is known.
    NOTE: will not refresh the cache itself to determine this.

    :param guid: Atlan-internal UUID of the object
    :returns: `True` if the object is known, `False` otherwise
    """
    return guid in self.guid_to_asset
is_name_known(name: str)

Checks whether the provided Atlan-internal ID string is known. NOTE: will not refresh the cache itself to determine this.

:param name: human-constructable name of the object :returns: True if the object is known, False otherwise

Source code in pyatlan/cache/abstract_asset_cache.py 
66
67
68
69
70
71
72
73
74
def is_name_known(self, name: str):
    """
    Checks whether the provided Atlan-internal ID string is known.
    NOTE: will not refresh the cache itself to determine this.

    :param name: human-constructable name of the object
    :returns: `True` if the object is known, `False` otherwise
    """
    return name in self.name_to_guid
is_qualified_name_known(qualified_name: str)

Checks whether the provided Atlan-internal ID string is known. NOTE: will not refresh the cache itself to determine this.

:param qualified_name: Atlan-internal ID string of the object :returns: True if the object is known, False otherwise

Source code in pyatlan/cache/abstract_asset_cache.py 
56
57
58
59
60
61
62
63
64
def is_qualified_name_known(self, qualified_name: str):
    """
    Checks whether the provided Atlan-internal ID string is known.
    NOTE: will not refresh the cache itself to determine this.

    :param qualified_name: Atlan-internal ID string of the object
    :returns: `True` if the object is known, `False` otherwise
    """
    return qualified_name in self.qualified_name_to_guid
lookup_by_guid(guid: str) abstractmethod

Abstract method to lookup asset by guid.

Source code in pyatlan/cache/abstract_asset_cache.py 
30
31
32
@abstractmethod
def lookup_by_guid(self, guid: str):
    """Abstract method to lookup asset by guid."""
lookup_by_name(name: Any) abstractmethod

Abstract method to lookup asset by name.

Source code in pyatlan/cache/abstract_asset_cache.py 
38
39
40
@abstractmethod
def lookup_by_name(self, name: Any):
    """Abstract method to lookup asset by name."""
lookup_by_qualified_name(qualified_name: str) abstractmethod

Abstract method to lookup asset by qualified name.

Source code in pyatlan/cache/abstract_asset_cache.py 
34
35
36
@abstractmethod
def lookup_by_qualified_name(self, qualified_name: str):
    """Abstract method to lookup asset by qualified name."""

AbstractAssetName()

Bases: ABC

Base class for reusable components common to all asset names used by the cache's find methods, such as AssetCache.get_by_name().

Source code in pyatlan/cache/abstract_asset_cache.py 
168
169
170
@abstractmethod
def __init__(self):
    pass

pyatlan.cache.dq_template_config_cache

Classes

DQTemplateConfigCache(client: AtlanClient)

Lazily-loaded cache for DQ rule template configurations to avoid multiple API calls.

Source code in pyatlan/cache/dq_template_config_cache.py 
19
20
21
22
23
def __init__(self, client: AtlanClient):
    self.client: AtlanClient = client
    self._cache: Dict[str, Dict] = {}
    self._lock: threading.Lock = threading.Lock()
    self._initialized: bool = False
Functions
get_template_config(rule_type: str) -> Optional[Dict]

Get template configuration for a specific rule type.

:param rule_type: The name of the rule template :returns: Template configuration dict or None if not found

Source code in pyatlan/cache/dq_template_config_cache.py 
31
32
33
34
35
36
37
38
39
40
41
def get_template_config(self, rule_type: str) -> Optional[Dict]:
    """
    Get template configuration for a specific rule type.

    :param rule_type: The name of the rule template
    :returns: Template configuration dict or None if not found
    """
    if not self._initialized:
        self._refresh_cache()

    return self._cache.get(rule_type)
refresh_cache() -> None

Refreshes the cache of DQ template configurations by requesting the full set from Atlan.

Source code in pyatlan/cache/dq_template_config_cache.py 
25
26
27
28
29
def refresh_cache(self) -> None:
    """
    Refreshes the cache of DQ template configurations by requesting the full set from Atlan.
    """
    self._refresh_cache()