Atlan Client Interface#

All the methods you can use to interact with Atlan using pyatlan.

class pyatlan.client.atlan.AtlanClient(*, base_url: Literal['INTERNAL'] | HttpUrl, api_key: str)[source]#

add_api_token_as_admin(asset_guid: str, impersonation_token: str) → AssetMutationResponse | None[source]#: Deprecated - use user.add_as_admin() instead.

add_api_token_as_viewer(asset_guid: str, impersonation_token: str) → AssetMutationResponse | None[source]#: Deprecated - use user.add_as_viewer() instead.

add_atlan_tags(asset_type: Type[A], qualified_name: str, atlan_tag_names: List[str], propagate: bool = True, remove_propagation_on_delete: bool = True, restrict_lineage_propagation: bool = True, restrict_propagation_through_hierarchy: bool = False) → None[source]#: Deprecated - use asset.add_atlan_tags() instead.

add_user_to_groups(guid: str, group_ids: List[str]) → None[source]#: Deprecated - use user.add_to_groups() instead.

append_terms(asset_type: Type[A], terms: List[AtlasGlossaryTerm], guid: str | None = None, qualified_name: str | None = None) → A[source]#: Deprecated - use asset.append_terms() instead.

change_user_role(guid: str, role_id: str) → None[source]#: Deprecated - use user.change_role() instead.

classmethod construct(_fields_set: SetStr | None = None, **values: Any) → Model#: Creates a new model setting __dict__ and __fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. Behaves as if Config.extra = ‘allow’ was set since it adds all passed values

Duplicate a model, optionally choose which fields to include, exclude and change.

Parameters:

include – fields to include in new model
exclude – fields to exclude from new model, as with values this takes precedence over include
update – values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data
deep – set to True to make a deep copy of the model

Returns:

new model instance

create_api_token(display_name: str, description: str = '', personas: Set[str] | None = None, validity_seconds: int = -1) → ApiToken[source]#: Deprecated - use token.create() instead.

create_group(group: AtlanGroup, user_ids: List[str] | None = None) → CreateGroupResponse[source]#: Deprecated - use group.create() instead.

create_typedef(typedef: TypeDef) → TypeDefResponse[source]#: Deprecated - use typedef.create() instead.

create_users(users: List[AtlanUser]) → None[source]#: Deprecated - use user.create() instead.

delete_entity_by_guid(guid: str | List[str]) → AssetMutationResponse[source]#: Deprecated - use asset.delete_by_guid() instead.

dict(*, include: AbstractSetIntStr | MappingIntStrAny | None = None, exclude: AbstractSetIntStr | MappingIntStrAny | None = None, by_alias: bool = False, skip_defaults: bool | None = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False) → DictStrAny#: Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

find_category_by_name(name: ConstrainedStrValue, glossary_name: ConstrainedStrValue, attributes: List[StrictStr] | None = None) → List[AtlasGlossaryCategory][source]#: Deprecated - use asset.find_category_by_name() instead.

find_category_fast_by_name(name: ConstrainedStrValue, glossary_qualified_name: ConstrainedStrValue, attributes: List[StrictStr] | None = None) → List[AtlasGlossaryCategory][source]#: Deprecated - use asset.find_category_fast_by_name() instead.

find_connections_by_name(name: str, connector_type: AtlanConnectorType, attributes: List[str] | None = None) → List[Connection][source]#: Deprecated - use asset.find_connections_by_name() instead.

find_glossary_by_name(name: ConstrainedStrValue, attributes: List[StrictStr] | None = None) → AtlasGlossary[source]#: Deprecated - use asset.find_glossary_by_name() instead.

find_personas_by_name(name: str, attributes: List[str] | None = None) → List[Persona][source]#: Deprecated - use asset.find_personas_by_name() instead.

find_purposes_by_name(name: str, attributes: List[str] | None = None) → List[Purpose][source]#: Deprecated - use asset.find_personas_by_name() instead.

find_term_by_name(name: ConstrainedStrValue, glossary_name: ConstrainedStrValue, attributes: List[StrictStr] | None = None) → AtlasGlossaryTerm[source]#: Deprecated - use asset.find_term_by_name() instead.

find_term_fast_by_name(name: ConstrainedStrValue, glossary_qualified_name: ConstrainedStrValue, attributes: List[StrictStr] | None = None) → AtlasGlossaryTerm[source]#: Deprecated - use asset.find_category_by_name() instead.

get_admin_events(admin_request: AdminEventRequest) → AdminEventResponse[source]#: Deprecated - use admin.get_admin_events() instead.

get_all_groups(limit: int = 20) → List[AtlanGroup][source]#: Deprecated - use group.get_all() instead.

get_all_roles() → RoleResponse[source]#: Deprecated - use self.role.get_all() instead.

get_all_typedefs() → TypeDefResponse[source]#: Deprecated - use typedef.get_all() instead.

get_all_users(limit: int = 20) → List[AtlanUser][source]#: Deprecated - use user.get_all() instead.

get_api_token_by_id(client_id: str) → ApiToken | None[source]#: Deprecated - use token.get_by_id() instead.

get_api_token_by_name(display_name: str) → ApiToken | None[source]#: Deprecated - use token.get_by_name() instead.

get_api_tokens(limit: int | None = None, post_filter: str | None = None, sort: str | None = None, count: bool = True, offset: int = 0) → ApiTokenResponse[source]#: Deprecated - use token.get() instead.

get_asset_by_guid(guid: str, asset_type: Type[A], min_ext_info: bool = False, ignore_relationships: bool = False) → A[source]#: Deprecated - use asset.get_by_guid() instead.

get_asset_by_qualified_name(qualified_name: str, asset_type: Type[A], min_ext_info: bool = False, ignore_relationships: bool = False) → A[source]#: Deprecated - use asset.get_by_qualified_name() instead.

get_current_user() → UserMinimalResponse[source]#: Deprecated - use user.get_current() instead.

classmethod get_default_client() → AtlanClient[source]#

Retrieves the default client.

Returns:: the default client

get_group_by_name(alias: str, limit: int = 20) → List[AtlanGroup] | None[source]#: Deprecated - use group.get_by_name() instead.

get_group_members(guid: str) → UserResponse[source]#: Deprecated - use group.get_members() instead.

get_groups(limit: int | None = None, post_filter: str | None = None, sort: str | None = None, count: bool = True, offset: int = 0) → GroupResponse[source]#: Deprecated - use group.get() instead.

get_groups_for_user(guid: str) → GroupResponse[source]#: Deprecated - use user.get_groups() instead.

get_keycloak_events(keycloak_request: KeycloakEventRequest) → KeycloakEventResponse[source]#: Deprecated - use admin.get_keycloak_events() instead.

get_lineage(lineage_request: LineageRequest) → LineageResponse[source]#

Deprecated — this is an older, slower operation to retrieve lineage that will not receive further enhancements. Use the get_lineage_list operation instead.

Parameters:: lineage_request – detailing the lineage query, parameters, and so on to run
Returns:: the results of the lineage request
Raises:: AtlanError – on any API communication issue

get_lineage_list(lineage_request: LineageListRequest) → LineageListResults[source]#: Deprecated - use asset.get_lineage_list() instead.

get_roles(limit: int, post_filter: str | None = None, sort: str | None = None, count: bool = True, offset: int = 0) → RoleResponse[source]#: Deprecated - use role.get() instead.

get_typedefs(type_category: AtlanTypeCategory | List[AtlanTypeCategory]) → TypeDefResponse[source]#: Deprecated - use typedef.get() instead.

get_user_by_username(username: str) → AtlanUser | None[source]#: Deprecated - use user.get_by_username() instead.

get_users(limit: int | None = None, post_filter: str | None = None, sort: str | None = None, count: bool = True, offset: int = 0) → UserResponse[source]#: Deprecated - use user.get() instead.

get_users_by_email(email: str, limit: int = 20) → List[AtlanUser] | None[source]#: Deprecated - use user.get_by_email() instead.

json(*, include: AbstractSetIntStr | MappingIntStrAny | None = None, exclude: AbstractSetIntStr | MappingIntStrAny | None = None, by_alias: bool = False, skip_defaults: bool | None = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, encoder: Callable[[Any], Any] | None = None, models_as_dict: bool = True, **dumps_kwargs: Any) → str#

Generate a JSON representation of the model, include and exclude arguments as per dict().

encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().

max_retries(max_retries: Retry = Retry(total=10, connect=None, read=None, redirect=None, status=None)) → Generator[None, None, None][source]#: Creates a context manger that can used to temporarily change parameters used for retrying connnections. The original Retry information will be restored when the context is exited.

parse_query(query: QueryParserRequest) → ParsedQuery | None[source]#

Parses the provided query to describe its component parts.

Parameters:: query – query to parse and configuration options
Returns:: parsed explanation of the query
Raises:: AtlanError – on any API communication issue

purge_api_token(guid: str) → None[source]#: Deprecated - use token.purge() instead.

purge_entity_by_guid(guid: str | List[str]) → AssetMutationResponse[source]#: Deprecated - use asset.purge_by_guid() instead.

purge_group(guid: str) → None[source]#: Deprecated - use group.purge() instead.

purge_typedef(name: str, typedef_type: type) → None[source]#: Deprecated - use typedef.purge() instead.

remove_announcement(asset_type: Type[A], qualified_name: str, name: str) → A | None[source]#: Deprecated - use asset.remove_announcement() instead.

remove_atlan_tag(asset_type: Type[A], qualified_name: str, atlan_tag_name: str) → None[source]#: Deprecated - use asset.remove_atlan_tag() instead.

remove_certificate(asset_type: Type[A], qualified_name: str, name: str) → A | None[source]#: Deprecated - use asset.remove_certificate() instead.

remove_custom_metadata(guid: str, cm_name: str)[source]#: Deprecated - use asset.remove_custom_metadata() instead.

remove_terms(asset_type: Type[A], terms: List[AtlasGlossaryTerm], guid: str | None = None, qualified_name: str | None = None) → A[source]#: Deprecated - use asset.remove_terms() instead.

remove_users_from_group(guid: str, user_ids=typing.List[str]) → None[source]#: Deprecated - use group.remove_users() instead.

replace_custom_metadata(guid: str, custom_metadata: CustomMetadataDict)[source]#: Deprecated - use asset.replace_custom_metadata() instead.

replace_terms(asset_type: Type[A], terms: List[AtlasGlossaryTerm], guid: str | None = None, qualified_name: str | None = None) → A[source]#: Deprecated - use asset.replace_terms() instead.

restore(asset_type: Type[A], qualified_name: str) → bool[source]#: Deprecated - use asset.restore() instead.

retrieve_minimal(guid: str, asset_type: Type[A]) → A[source]#: Deprecated - use asset.retrieve_minimal() instead.

save(entity: Asset | List[Asset], replace_atlan_tags: bool = False, replace_custom_metadata: bool = False, overwrite_custom_metadata: bool = False) → AssetMutationResponse[source]#: Deprecated - use asset.save() instead.

save_merging_cm(entity: Asset | List[Asset], replace_atlan_tags: bool = False) → AssetMutationResponse[source]#: Deprecated - use asset.save_merging_cm() instead.

save_replacing_cm(entity: Asset | List[Asset], replace_atlan_tags: bool = False) → AssetMutationResponse[source]#: Deprecated - use asset.save_replacing_cm() instead.

search(criteria: IndexSearchRequest) → IndexSearchResults[source]#: Deprecated - use asset.search() instead.

classmethod set_default_client(client: AtlanClient)[source]#: Sets the default client to be used by caches

update_announcement(asset_type: Type[A], qualified_name: str, name: str, announcement: Announcement) → A | None[source]#: Deprecated - use asset.update_announcement() instead.

update_api_token(guid: str, display_name: str, description: str = '', personas: Set[str] | None = None) → ApiToken[source]#: Deprecated - use token.update() instead.

update_certificate(asset_type: Type[A], qualified_name: str, name: str, certificate_status: CertificateStatus, message: str | None = None) → A | None[source]#: Deprecated - use asset.update_certificate() instead.

update_custom_metadata_attributes(guid: str, custom_metadata: CustomMetadataDict)[source]#: Deprecated - use asset.update_custom_metadata_attributes() instead.

classmethod update_forward_refs(**localns: Any) → None#: Try to update ForwardRefs on fields based on this Model, globalns and localns.

update_group(group: AtlanGroup) → None[source]#: Deprecated - use group.update() instead.

update_merging_cm(entity: Asset, replace_atlan_tags: bool = False) → AssetMutationResponse[source]#: Deprecated - use asset.update_merging_cm() instead.

update_replacing_cm(entity: Asset, replace_atlan_tags: bool = False) → AssetMutationResponse[source]#: Deprecated - use asset.update_replacing_cm() instead.

update_typedef(typedef: TypeDef) → TypeDefResponse[source]#: Deprecated - use typedef.update() instead.

update_user(guid: str, user: AtlanUser) → UserMinimalResponse[source]#: Deprecated - use user.update() instead.

upload_image(file, filename: str) → AtlanImage[source]#

Uploads an image from the provided local file.

Parameters:

file – local file to upload
filename – name of the file to be uploaded

Returns:

details of the uploaded image

Raises:

AtlanError – on any API communication issue

upsert(entity: Asset | List[Asset], replace_atlan_tags: bool = False, replace_custom_metadata: bool = False, overwrite_custom_metadata: bool = False) → AssetMutationResponse[source]#: Deprecated - use asset.save() instead.

upsert_merging_cm(entity: Asset | List[Asset], replace_atlan_tags: bool = False) → AssetMutationResponse[source]#: Deprecated - use asset.save_merging_cm() instead.

upsert_replacing_cm(entity: Asset | List[Asset], replace_atlan_tagss: bool = False) → AssetMutationResponse[source]#: Deprecated - use asset.save_replacing_cm() instead.