content: # Copyright Contributors to the Amundsen project.
  # SPDX-License-Identifier: Apache-2.0
  
  import datetime
  import logging
  import re
  from collections import defaultdict
  from operator import attrgetter
  from random import randint
  from typing import (Any, Dict, Generator, List, Optional, Set, Tuple, Type,
                      Union, no_type_check)
  
  from amundsen_common.entity.resource_type import ResourceType
  from amundsen_common.models.dashboard import DashboardSummary
  from amundsen_common.models.feature import Feature
  from amundsen_common.models.generation_code import GenerationCode
  from amundsen_common.models.lineage import Lineage, LineageItem
  from amundsen_common.models.popular_table import PopularTable
  from amundsen_common.models.table import (Application, Badge, Column,
                                            ProgrammaticDescription, Reader,
                                            ResourceReport, Stat, Table, Tag,
                                            User, Watermark)
  from amundsen_common.models.user import User as UserEntity
  from amundsen_common.utils.atlas import (AtlasColumnKey, AtlasCommonParams,
                                           AtlasCommonTypes, AtlasDashboardTypes,
                                           AtlasStatus, AtlasTableKey,
                                           AtlasTableTypes)
  from apache_atlas.client.base_client import AtlasClient
  from apache_atlas.model.glossary import (AtlasGlossary, AtlasGlossaryHeader,
                                           AtlasGlossaryTerm)
  from apache_atlas.model.instance import (AtlasEntitiesWithExtInfo, AtlasEntity,
                                           AtlasEntityHeader,
                                           AtlasEntityWithExtInfo,
                                           AtlasRelatedObjectId)
  from apache_atlas.model.relationship import AtlasRelationship
  from apache_atlas.utils import type_coerce
  from beaker.cache import CacheManager
  from beaker.util import parse_cache_config_options
  from flask import current_app as app
  from werkzeug.exceptions import BadRequest
  
  from metadata_service.entity.dashboard_detail import \
      DashboardDetail as DashboardDetailEntity
  from metadata_service.entity.dashboard_query import DashboardQuery
  from metadata_service.entity.description import Description
  from metadata_service.entity.tag_detail import TagDetail
  from metadata_service.exception import NotFoundException
  from metadata_service.proxy import BaseProxy
  from metadata_service.util import UserResourceRel
  
  LOGGER = logging.getLogger(__name__)
  
  # Expire cache every 11 hours + jitter
  _ATLAS_PROXY_CACHE_EXPIRY_SEC = 11 * 60 * 60 + randint(0, 3600)
  
  
  # noinspection PyMethodMayBeStatic
  class AtlasProxy(BaseProxy):
      """
      Atlas Proxy client for the amundsen metadata
      {ATLAS_API_DOCS} = https://atlas.apache.org/api/v2/
      """
      DB_ATTRIBUTE = 'db'
      STATISTICS_FORMAT_SPEC = app.config['STATISTICS_FORMAT_SPEC']
      # Qualified Name of the Glossary, that holds the user defined terms.
      # For Amundsen, we are using Glossary Terms as the Tags.
      AMUNDSEN_USER_TAGS = 'amundsen_user_tags'
      _CACHE = CacheManager(**parse_cache_config_options({'cache.regions': 'atlas_proxy',
                                                          'cache.atlas_proxy.type': 'memory',
                                                          'cache.atlas_proxy.expire': _ATLAS_PROXY_CACHE_EXPIRY_SEC}))
  
      def __init__(self, *,
                   host: str,
                   port: int,
                   user: str = 'admin',
                   password: str = '',
                   encrypted: bool = False,
                   validate_ssl: bool = False,
                   client_kwargs: dict = dict(),
                   **kwargs: dict) -> None:
          """
          Initiate the Apache Atlas client with the provided credentials
          """
          protocol = 'https' if encrypted else 'http'
          self.client = AtlasClient(f'{protocol}://{host}:{port}', (user, password))
          self.client.session.verify = validate_ssl
  
      def _parse_dashboard_bookmark_qn(self, bookmark_qn: str) -> Dict:
          """
          Parse bookmark qualifiedName and extract the info
          :param bookmark_qn: Qualified Name of Bookmark entity
          :return: Dictionary object containing following information:
          product: dashboard product
          cluster: cluster information
          dashboard_group: Dashboard group name
          dashboard_id: Dashboard identifier
          user_id: User id
          """
          pattern = re.compile(r"""
          ^(?P<product>[^.]*)_dashboard
          ://
          (?P<cluster>[^.]*)
          \.
          (?P<dashboard_group>[^.]*)
          /
          (?P<dashboard_id>[^.]*)
          /
          (?P<type>[^.]*)
          /
          bookmark
          /
          (?P<user_id>[^.]*)
          $
          """, re.X)
          result = pattern.match(bookmark_qn)
          return result.groupdict() if result else dict()
  
      def _parse_table_bookmark_qn(self, bookmark_qn: str) -> Dict:
          """
          Parse bookmark qualifiedName and extract the info
          :param bookmark_qn: Qualified Name of Bookmark entity
          :return: Dictionary object containing following information:
          cluster: cluster information
          db: Database name
          name: Table name
          """
          pattern = re.compile(r"""
          ^(?P<db>[^.]*)
          \.
          (?P<table>[^.]*)
          \.
          (?P<entity_type>[^.]*)
          \.
          (?P<user_id>[^.]*)\.bookmark
          \@
          (?P<cluster>.*)
          $
          """, re.X)
          result = pattern.match(bookmark_qn)
          return result.groupdict() if result else dict()
  
      @classmethod
      def _filter_active(cls, entities: List[dict]) -> List[dict]:
          """
          Filter out active entities based on entity end relationship status.
          """
          result = [e for e in entities
                    if e.get('relationshipStatus') == AtlasStatus.ACTIVE
                    and e.get('entityStatus') == AtlasStatus.ACTIVE]
  
          return result
  
      def _get_table_entity(self, *, table_uri: str) -> AtlasEntityWithExtInfo:
          """
          Fetch information from table_uri and then find the appropriate entity
          :param table_uri: The table URI coming from Amundsen Frontend
          :return: A table entity matching the Qualified Name derived from table_uri
          """
          key = AtlasTableKey(table_uri)
  
          try:
              return self.client.entity.get_entity_by_attribute(type_name=key.entity_type,
                                                                uniq_attributes=[
                                                                    (AtlasCommonParams.qualified_name,
                                                                     key.qualified_name)])
          except Exception as ex:
              LOGGER.exception(f'Table not found. {str(ex)}')
              raise NotFoundException(f'Table URI( {table_uri} ) does not exist')
  
      def _get_user_entity(self, user_id: str) -> AtlasEntityWithExtInfo:
          """
          Fetches an user entity from an id
          :param user_id: User ID
          :return: A User entity matching the user_id
          """
          try:
              return self.client.entity.get_entity_by_attribute(type_name=AtlasCommonTypes.user,
                                                                uniq_attributes=[
                                                                    (AtlasCommonParams.qualified_name, user_id)])
          except Exception:
              raise NotFoundException(f'(User {user_id}) does not exist')
  
      def _create_bookmark(self, entity: AtlasEntityWithExtInfo, user_guid: str, bookmark_qn: str,
                           entity_uri: str) -> None:
          """
          Creates a bookmark entity for a specific user and entity uri.
          :param entity: bookmarked entity
          :param user_guid: User's guid
          :param bookmark_qn: Bookmark qualifiedName
          :param entity_uri: uri of bookmarked entity
          :return:
          """
  
          bookmark_entity = {
              'entity': {
                  'typeName': AtlasCommonTypes.bookmark,
                  AtlasCommonParams.attributes: {
                      AtlasCommonParams.qualified_name: bookmark_qn,
                      AtlasStatus.ACTIVE.lower(): True,
                      'entityUri': entity_uri,
                      'entityName': entity.entity[AtlasCommonParams.attributes]['name'],
                      'user': {AtlasCommonParams.guid: user_guid},
                      'entity': {AtlasCommonParams.guid: entity.entity[AtlasCommonParams.guid]}}
              }
          }
  
          bookmark_entity = type_coerce(bookmark_entity, AtlasEntityWithExtInfo)
          self.client.entity.create_entity(bookmark_entity)
  
      def _get_bookmark_entity(self, entity_uri: str, user_id: str,
                               resource_type: ResourceType) -> AtlasEntityWithExtInfo:
          """
          Fetch a Bookmark entity from parsing entity uri and user id.
          If Bookmark is not present, create one for the user.
          :param entity_uri:
          :param user_id: Qualified Name of a user
          :return:
          """
          if resource_type == ResourceType.Table:
              entity_info = AtlasTableKey(entity_uri).get_details()
  
              schema = entity_info.get('schema')
              table = entity_info.get('table')
              database = entity_info.get('database', 'hive_table')
              cluster = entity_info.get('cluster')
  
              bookmark_qn = f'{schema}.{table}.{database}.{user_id}.bookmark@{cluster}'
  
          else:
              bookmark_qn = f'{entity_uri}/{resource_type.name.lower()}/bookmark/{user_id}'
  
          try:
              bookmark_entity = self.client.entity.get_entity_by_attribute(type_name=AtlasCommonTypes.bookmark,
                                                                           uniq_attributes=[
                                                                               (AtlasCommonParams.qualified_name,
                                                                                bookmark_qn)])
          except Exception as ex:
              LOGGER.exception(f'Bookmark not found. {str(ex)}')
  
              if resource_type == ResourceType.Table:
                  bookmarked_entity = self._get_table_entity(table_uri=entity_uri)
              elif resource_type == ResourceType.Dashboard:
                  bookmarked_entity = self._get_dashboard(qualified_name=entity_uri)
              else:
                  raise NotImplementedError(f'Bookmarks for Resource Type ({resource_type}) are not yet implemented')
  
              # Fetch user entity from user_id for relation
              user_entity = self._get_user_entity(user_id)
              # Create bookmark entity with the user relation.
  
              self._create_bookmark(bookmarked_entity,
                                    user_entity.entity[AtlasCommonParams.guid],
                                    bookmark_qn,
                                    entity_uri)
              # Fetch bookmark entity after creating it.
              bookmark_entity = self.client.entity.get_entity_by_attribute(type_name=AtlasCommonTypes.bookmark,
                                                                           uniq_attributes=[
                                                                               (AtlasCommonParams.qualified_name,
                                                                                bookmark_qn)])
  
          return bookmark_entity
  
      def _get_column(self, *, table_uri: str, column_name: str) -> Dict:
          """
          Fetch the column information from referredEntities of the table entity
          :param table_uri:
          :param column_name:
          :return: A dictionary containing the column details
          """
          try:
              table_entity = self._get_table_entity(table_uri=table_uri)
              columns = table_entity.entity[AtlasCommonParams.relationships].get('columns')
              for column in columns or list():
                  col_details = table_entity.referredEntities[column[AtlasCommonParams.guid]]
                  if column_name == col_details[AtlasCommonParams.attributes]['name']:
                      return col_details
  
              raise NotFoundException(f'Column not found: {column_name}')
  
          except KeyError as ex:
              LOGGER.exception(f'Column not found: {str(ex)}')
              raise NotFoundException(f'Column not found: {column_name}')
  
      def _serialize_columns(self, *, entity: AtlasEntityWithExtInfo) -> \
              Union[List[Column], List]:
          """
          Helper function to fetch the columns from entity and serialize them
          using Column and Stat model.
          :param entity: AtlasEntityWithExtInfo object,
          along with relationshipAttributes
          :return: A list of Column objects, if there are any columns available,
          else an empty list.
          """
          columns = list()
          for column in entity.entity[AtlasCommonParams.relationships].get('columns') or list():
              column_status = column.get('entityStatus', 'inactive').lower()
  
              if column_status != 'active':
                  continue
  
              col_entity = entity.referredEntities[column[AtlasCommonParams.guid]]
              col_attrs = col_entity[AtlasCommonParams.attributes]
              statistics = list()
  
              badges = list()
              for column_classification in col_entity.get('classifications') or list():
                  if column_classification.get('entityStatus') == AtlasStatus.ACTIVE:
                      name = column_classification.get('typeName')
  
                      badges.append(Badge(badge_name=name, category='default'))
  
              for stats in col_attrs.get('statistics') or list():
                  stats_attrs = stats[AtlasCommonParams.attributes]
  
                  stat_type = stats_attrs.get('stat_name')
  
                  stat_format = self.STATISTICS_FORMAT_SPEC.get(stat_type, dict())
  
                  if not stat_format.get('drop', False):
                      stat_type = stat_format.get('new_name', stat_type)
  
                      stat_val = stats_attrs.get('stat_val')
  
                      format_val = stat_format.get('format')
  
                      if format_val:
                          stat_val = format_val.format(stat_val)
                      else:
                          stat_val = str(stat_val)
  
                      start_epoch = stats_attrs.get('start_epoch')
                      end_epoch = stats_attrs.get('end_epoch')
  
                      statistics.append(
                          Stat(
                              stat_type=stat_type,
                              stat_val=stat_val,
                              start_epoch=start_epoch,
                              end_epoch=end_epoch,
                          )
                      )
  
              columns.append(
                  Column(
                      name=col_attrs.get('name'),
                      description=col_attrs.get('description') or col_attrs.get('comment'),
                      col_type=col_attrs.get('type') or col_attrs.get('dataType') or col_attrs.get('data_type'),
                      sort_order=col_attrs.get('position') or 9999,
                      stats=statistics,
                      badges=badges
                  )
              )
          return sorted(columns, key=lambda item: item.sort_order)
  
      def _get_reports(self, guids: List[str]) -> List[ResourceReport]:
          reports = []
          if guids:
              report_entities = self.client.entity.get_entities_by_guids(guids=guids)
              for report_entity in report_entities.entities:
                  try:
                      if report_entity.status == AtlasStatus.ACTIVE:
                          report_attrs = report_entity.attributes
                          reports.append(
                              ResourceReport(
                                  name=report_attrs['name'],
                                  url=report_attrs['url']
                              )
                          )
                  except (KeyError, AttributeError):
                      LOGGER.exception(f'Error while accessing table report: {str(report_entity)}', exc_info=True)
  
          parsed_reports = app.config['RESOURCE_REPORT_CLIENT'](reports) \
              if app.config['RESOURCE_REPORT_CLIENT'] else reports
  
          return sorted(parsed_reports)
  
      def _get_owners(self, data_owners: list, fallback_owner: Optional[str] = None) -> List[User]:
          owners_detail = list()
          active_owners_list = list()
  
          for owner in self._filter_active(data_owners):
              owner_qn = owner['displayText']
              owner_data = self._get_user_details(owner_qn)
              owners_detail.append(User(**owner_data))
              active_owners_list.append(owner_qn)
  
          # To avoid the duplication,
          # we are checking if the fallback is not in data_owners
          if fallback_owner and (fallback_owner not in active_owners_list):
              owners_detail.append(User(**self._get_user_details(fallback_owner)))
  
          return owners_detail
  
      def get_user(self, *, id: str) -> Union[UserEntity, None]:
          pass
  
      @no_type_check
      def create_update_user(self, *, user: User) -> Tuple[User, bool]:
          pass
  
      @no_type_check
      def get_users(self) -> List[UserEntity]:
          pass
  
      def _serialize_badges(self, entity: AtlasEntityWithExtInfo) -> List[Badge]:
          """
          Return list of Badges for entity. Badges in Amundsen <> Atlas integration are based on Atlas Classification.
  
          :param entity: entity for which badges should be collected
          :return : List of Amundsen Badge objects.
          """
          result = []
  
          classifications = entity.get('classifications')
  
          for classification in classifications or list():
              result.append(Badge(badge_name=classification.get('typeName'), category='default'))
  
          return result
  
      def _serialize_tags(self, entity: AtlasEntityWithExtInfo) -> List[Tag]:
          """
          Return list of Tags for entity. Tags in Amundsen <> Atlas integration are based on Atlas Glossary.
  
          :param entity: entity for which tags should be collected
          :return : List of Amundsen Tag objects.
          """
          result = []
  
          meanings = self._filter_active(entity.get(AtlasCommonParams.relationships, dict()).get('meanings', []))
  
          for term in meanings or list():
              result.append(Tag(tag_name=term.get('displayText', ''), tag_type='default'))
  
          return result
  
      def get_table(self, *, table_uri: str) -> Table:
          """
          Gathers all the information needed for the Table Detail Page.
          :param table_uri:
          :return: A Table object with all the information available
          or gathered from different entities.
          """
          entity = self._get_table_entity(table_uri=table_uri)
          table_details = entity.entity
  
          try:
              attrs = table_details[AtlasCommonParams.attributes]
  
              programmatic_descriptions = self._get_programmatic_descriptions(attrs.get('parameters', dict()) or dict())
  
              table_info = AtlasTableKey(attrs.get(AtlasCommonParams.qualified_name)).get_details()
  
              badges = self._serialize_badges(table_details)
              tags = self._serialize_tags(table_details)
  
              columns = self._serialize_columns(entity=entity)
  
              reports_guids = [report.get("guid") for report in attrs.get("reports") or list()]
  
              table_type = attrs.get('tableType') or 'table'
              is_view = 'view' in table_type.lower()
  
              readers = self._get_readers(table_details, Reader)
              application = self._get_application(table_details)
  
              table = Table(
                  table_writer=application,
                  database=AtlasTableKey(table_uri).get_details()['database'],
                  cluster=table_info.get('cluster', ''),
                  schema=table_info.get('schema', ''),
                  name=attrs.get('name') or table_info.get('table', ''),
                  badges=badges,
                  tags=tags,
                  description=attrs.get('description') or attrs.get('comment'),
                  owners=self._get_owners(
                      table_details[AtlasCommonParams.relationships].get('ownedBy', []), attrs.get('owner')),
                  resource_reports=self._get_reports(guids=reports_guids),
                  columns=columns,
                  is_view=is_view,
                  table_readers=readers,
                  last_updated_timestamp=self._parse_date(table_details.get('updateTime')),
                  programmatic_descriptions=programmatic_descriptions,
                  watermarks=self._get_table_watermarks(table_details))
  
              return table
          except KeyError:
              LOGGER.exception('Error while accessing table information. {}', exc_info=True)
              raise BadRequest(f'Some of the required attributes are missing in: {table_uri}')
  
      @staticmethod
      def _validate_date(text_date: str, date_format: str) -> Tuple[Optional[datetime.datetime], Optional[str]]:
          try:
              return datetime.datetime.strptime(text_date, date_format), date_format
          except (ValueError, TypeError):
              return None, None
  
      @staticmethod
      def _select_watermark_format(partition_names: Optional[List[Any]]) -> Optional[str]:
          result = None
  
          if partition_names:
              for partition_name in partition_names:
                  # Assume that all partitions for given table have the same date format. Only thing that needs to be done
                  # is establishing which format out of the supported ones it is and then we validate every partition
                  # against it.
                  for df in app.config['WATERMARK_DATE_FORMATS']:
                      _, result = AtlasProxy._validate_date(partition_name, df)
  
                      if result:
                          LOGGER.debug('Established date format', extra=dict(date_format=result))
                          return result
  
          return result
  
      @staticmethod
      def _render_partition_key_name(entity: AtlasEntityWithExtInfo) -> Optional[str]:
          _partition_keys = []
  
          for partition_key in entity.get(AtlasCommonParams.attributes, dict()).get('partitionKeys', []):
              partition_key_column_name = partition_key.get('displayName')
  
              if partition_key_column_name:
                  _partition_keys.append(partition_key_column_name)
  
          partition_key = ' '.join(_partition_keys).strip()
  
          return partition_key
  
      def _get_table_watermarks(self, entity: AtlasEntityWithExtInfo) -> List[Watermark]:
          partition_value_format = '%Y-%m-%d %H:%M:%S'
  
          _partitions = entity.get(AtlasCommonParams.relationships, dict()).get('partitions', list())
  
          names = [_partition.get('displayText') for _partition in self._filter_active(_partitions)]
  
          if not names:
              return []
  
          partition_key = self._render_partition_key_name(entity)
          watermark_date_format = self._select_watermark_format(names)
  
          partitions = {}
  
          for _partition in _partitions:
              partition_name = _partition.get('displayText')
              if partition_name and watermark_date_format:
                  partition_date, _ = self._validate_date(partition_name, watermark_date_format)
  
                  if partition_date:
                      common_values = {'partition_value': datetime.datetime.strftime(partition_date,
                                                                                     partition_value_format),
                                       'create_time': 0,
                                       'partition_key': partition_key}
  
                      partitions[partition_date] = common_values
  
          if partitions:
              low_watermark_date = min(partitions.keys())
              high_watermark_date = max(partitions.keys())
  
              low_watermark = Watermark(watermark_type='low_watermark', **partitions.get(low_watermark_date))
              high_watermark = Watermark(watermark_type='high_watermark', **partitions.get(high_watermark_date))
  
              return [low_watermark, high_watermark]
          else:
              return []
  
      def delete_owner(self, *, table_uri: str, owner: str) -> None:
          """
          :param table_uri:
          :param owner:
          :return:
          """
          table = self._get_table_entity(table_uri=table_uri)
          table_entity = table.entity
  
          if table_entity[AtlasCommonParams.relationships].get("ownedBy"):
              try:
                  active_owner = next(filter(lambda item:
                                             item['relationshipStatus'] == AtlasStatus.ACTIVE
                                             and item['displayText'] == owner,
                                             table_entity[AtlasCommonParams.relationships]['ownedBy']), None)
  
                  if active_owner:
                      self.client.relationship.delete_relationship_by_guid(
                          guid=active_owner.get('relationshipGuid')
                      )
                  else:
                      raise BadRequest('You can not delete this owner.')
              except Exception:
                  LOGGER.exception('Error while removing table data owner.', exc_info=True)
  
      def add_owner(self, *, table_uri: str, owner: str) -> None:
          """
          Query on Atlas User entity to find if the entity exist for the
          owner string in parameter, if not create one. And then use that User
          entity's GUID and add a relationship between Table and User, on ownedBy field.
          :param table_uri:
          :param owner: Email address of the owner
          :return: None, as it simply adds the owner.
          """
          owner_info = self._get_user_details(owner)
  
          if not owner_info:
              raise NotFoundException(f'User "{owner}" does not exist.')
  
          user_dict = type_coerce({
              "entity": {
                  "typeName": "User",
                  "attributes": {"qualifiedName": owner},
              }
          }, AtlasEntityWithExtInfo)
  
          # Get or Create a User
          user_entity = self.client.entity.create_entity(user_dict)
          user_guid = next(iter(user_entity.guidAssignments.values()))
  
          table = self._get_table_entity(table_uri=table_uri)
  
          entity_def = {
              "typeName": "DataSet_Users_Owner",
              "end1": {
                  "guid": table.entity.get("guid"), "typeName": "Table",
              },
              "end2": {
                  "guid": user_guid, "typeName": "User",
              },
          }
          try:
              relationship = type_coerce(entity_def, AtlasRelationship)
              self.client.relationship.create_relationship(relationship=relationship)
  
          except Exception:
              LOGGER.exception('Error while adding the owner information. {}', exc_info=True)
              raise BadRequest(f'User {owner} is already added as a data owner for table {table_uri}.')
  
      def get_table_description(self, *,
                                table_uri: str) -> Union[str, None]:
          """
          :param table_uri:
          :return: The description of the table as a string
          """
          entity = self._get_table_entity(table_uri=table_uri)
          return entity.entity[AtlasCommonParams.attributes].get('description')
  
      def put_table_description(self, *,
                                table_uri: str,
                                description: str) -> None:
          """
          Update the description of the given table.
          :param table_uri:
          :param description: Description string
          :return: None
          """
          table = self._get_table_entity(table_uri=table_uri)
  
          self.client.entity.partial_update_entity_by_guid(
              entity_guid=table.entity.get("guid"), attr_value=description, attr_name='description'
          )
  
      @_CACHE.cache('_get_user_defined_glossary_guid')
      def _get_user_defined_glossary_guid(self) -> str:
          """
          This function look for a user defined glossary i.e., self.ATLAS_USER_DEFINED_TERMS
          If there is not one available, this will create a new glossary.
          The main reason to put this functionality into a separate function is to avoid
          the lookup each time someone assigns a tag to a data source.
          :return: Glossary object, that holds the user defined terms.
          """
          # Check if the user glossary already exists
          glossaries = self.client.glossary.get_all_glossaries()
          for glossary in glossaries:
              if glossary.get(AtlasCommonParams.qualified_name) == self.AMUNDSEN_USER_TAGS:
                  return glossary[AtlasCommonParams.guid]
  
          # If not already exists, create one
          glossary_def = AtlasGlossary({"name": self.AMUNDSEN_USER_TAGS,
                                        "shortDescription": "Amundsen User Defined Terms"})
          glossary = self.client.glossary.create_glossary(glossary_def)
          return glossary.guid
  
      @_CACHE.cache('_get_create_glossary_term')
      def _get_create_glossary_term(self, term_name: str) -> Union[AtlasGlossaryTerm, AtlasEntityHeader]:
          """
          Since Atlas does not provide any API to find a term directly by a qualified name,
          we need to look for AtlasGlossaryTerm via basic search, if found then return, else
          create a new glossary term under the user defined glossary.
          :param term_name: Name of the term. NOTE: this is different from qualified name.
          :return: Term Object.
          """
          params = {
              'typeName': "AtlasGlossaryTerm",
              'excludeDeletedEntities': True,
              'includeSubTypes': True,
              AtlasCommonParams.attributes: ["assignedEntities", ],
              'entityFilters': {'condition': "AND",
                                'criterion': [{'attributeName': "name", 'operator': "=", 'attributeValue': term_name}]
                                }
          }
          result = self.client.discovery.faceted_search(search_parameters=params)
          if result.approximateCount:
              term = result.entities[0]
          else:
              glossary_guid = self._get_user_defined_glossary_guid()
              glossary_def = AtlasGlossaryHeader({'glossaryGuid': glossary_guid})
              term_def = AtlasGlossaryTerm({'name': term_name, 'anchor': glossary_def})
              term = self.client.glossary.create_glossary_term(term_def)
  
          return term
  
      def add_tag(self, *, id: str, tag: str, tag_type: str = "default",
                  resource_type: ResourceType = ResourceType.Table) -> None:
          """
          Assign the Glossary Term to the give table. If the term is not there, it will
          create a new term under the Glossary self.ATLAS_USER_DEFINED_TERMS
          :param id: Table URI / Dashboard ID etc.
          :param tag: Tag Name
          :param tag_type
          :return: None
          """
          entity = self._get_table_entity(table_uri=id)
  
          term = self._get_create_glossary_term(tag)
          related_entity = AtlasRelatedObjectId({AtlasCommonParams.guid: entity.entity[AtlasCommonParams.guid],
                                                 "typeName": resource_type.name})
          self.client.glossary.assign_term_to_entities(term.guid, [related_entity])
  
      def add_badge(self, *, id: str, badge_name: str, category: str = '',
                    resource_type: ResourceType) -> None:
          # Not implemented
          raise NotImplementedError
  
      def delete_tag(self, *, id: str, tag: str, tag_type: str,
                     resource_type: ResourceType = ResourceType.Table) -> None:
          """
          Removes the Glossary Term assignment from the provided source.
          :param id: Table URI / Dashboard ID etc.
          :param tag: Tag Name
          :return:None
          """
          entity = self._get_table_entity(table_uri=id)
          term = self._get_create_glossary_term(tag)
  
          if not term:
              return
  
          assigned_entities = self.client.glossary.get_entities_assigned_with_term(term.guid, "ASC", -1, 0)
  
          for item in assigned_entities or list():
              if item.get(AtlasCommonParams.guid) == entity.entity[AtlasCommonParams.guid]:
                  related_entity = AtlasRelatedObjectId(item)
                  return self.client.glossary.disassociate_term_from_entities(term.guid, [related_entity])
  
      def delete_badge(self, *, id: str, badge_name: str, category: str,
                       resource_type: ResourceType) -> None:
          # Not implemented
          raise NotImplementedError
  
      def put_column_description(self, *,
                                 table_uri: str,
                                 column_name: str,
  