content: # Copyright Contributors to the Amundsen project.
  # SPDX-License-Identifier: Apache-2.0
  
  import logging
  import time
  from collections import namedtuple
  from random import randint
  from typing import (Any, Dict, Iterator, List, Optional, Tuple, Type, Union,
                      no_type_check)
  
  from amundsen_common.entity.resource_type import ResourceType, to_resource_type
  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
  from amundsen_common.models.table import Badge
  from amundsen_common.models.table import Badge as TableBadge
  from amundsen_common.models.table import (Column, ProgrammaticDescription,
                                            Reader, Source, Stat, Table,
                                            TableSummary, Tag, User, Watermark)
  from amundsen_common.models.user import User as UserEntity
  from amundsen_common.models.user import UserSchema
  from amundsen_rds.models import RDSModel
  from amundsen_rds.models.badge import Badge as RDSBadge
  from amundsen_rds.models.base import Base
  from amundsen_rds.models.cluster import Cluster as RDSCluster
  from amundsen_rds.models.column import \
      ColumnDescription as RDSColumnDescription
  from amundsen_rds.models.column import TableColumn as RDSColumn
  from amundsen_rds.models.dashboard import Dashboard as RDSDashboard
  from amundsen_rds.models.dashboard import DashboardChart as RDSDashboardChart
  from amundsen_rds.models.dashboard import \
      DashboardCluster as RDSDashboardCluster
  from amundsen_rds.models.dashboard import \
      DashboardDescription as RDSDashboardDescription
  from amundsen_rds.models.dashboard import \
      DashboardExecution as RDSDashboardExecution
  from amundsen_rds.models.dashboard import \
      DashboardFollower as RDSDashboardFollower
  from amundsen_rds.models.dashboard import DashboardGroup as RDSDashboardGroup
  from amundsen_rds.models.dashboard import DashboardOwner as RDSDashboardOwner
  from amundsen_rds.models.dashboard import DashboardQuery as RDSDashboardQuery
  from amundsen_rds.models.dashboard import DashboardTable as RDSDashboardTable
  from amundsen_rds.models.dashboard import DashboardTag as RDSDashboardTag
  from amundsen_rds.models.dashboard import DashboardUsage as RDSDashboardUsage
  from amundsen_rds.models.database import Database as RDSDatabase
  from amundsen_rds.models.schema import Schema as RDSSchema
  from amundsen_rds.models.table import Table as RDSTable
  from amundsen_rds.models.table import TableDescription as RDSTableDescription
  from amundsen_rds.models.table import TableFollower as RDSTableFollower
  from amundsen_rds.models.table import TableOwner as RDSTableOwner
  from amundsen_rds.models.table import TableTag as RDSTableTag
  from amundsen_rds.models.table import TableUsage as RDSTableUsage
  from amundsen_rds.models.tag import Tag as RDSTag
  from amundsen_rds.models.updated_timestamp import \
      UpdatedTimestamp as RDSUpdatedTimestamp
  from amundsen_rds.models.user import User as RDSUser
  from beaker.cache import CacheManager
  from beaker.util import parse_cache_config_options
  from flask import current_app as app
  from sqlalchemy import func, literal
  from sqlalchemy.orm import Session, load_only, subqueryload
  
  from metadata_service.client.rds_client import RDSClient
  from metadata_service.entity.dashboard_detail import \
      DashboardDetail as DashboardDetailEntity
  from metadata_service.entity.dashboard_query import \
      DashboardQuery as DashboardQueryEntity
  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.base_proxy import BaseProxy
  from metadata_service.proxy.statsd_utilities import timer_with_counter
  from metadata_service.util import UserResourceRel
  
  _CACHE = CacheManager(**parse_cache_config_options({'cache.type': 'memory'}))
  
  # Expire cache every 11 hours + jitter
  _GET_POPULAR_RESOURCES_CACHE_EXPIRY_SEC = 11 * 60 * 60 + randint(0, 3600)
  
  resource_relation_model = {
      ResourceType.Table: {
          UserResourceRel.read: RDSTableUsage,
          UserResourceRel.own: RDSTableOwner,
          UserResourceRel.follow: RDSTableFollower
      },
      ResourceType.Dashboard: {
          UserResourceRel.read: RDSDashboardUsage,
          UserResourceRel.own: RDSDashboardOwner,
          UserResourceRel.follow: RDSDashboardFollower
      }
  }
  
  Edge = namedtuple('Edge', ['in_node', 'out_node'])
  EdgePair = namedtuple('EdgePair', ['in_edge', 'out_edge'])
  
  LOGGER = logging.getLogger(__name__)
  
  
  class MySQLProxy(BaseProxy):
      """
      A proxy to MySQL using SQLAlchemy ORM and Amundsen RDS
      See https://docs.sqlalchemy.org/en/13/orm/
      See https://github.com/amundsen-io/amundsenrds
      """
      def __init__(self, *,
                   host: Optional[str] = None,
                   port: Optional[int] = None,
                   user: Optional[str] = None,
                   password: Optional[str] = None,
                   client_kwargs: Dict = dict(),
                   **kwargs: Dict
                   ) -> None:
          endpoint = app.config['SQLALCHEMY_DATABASE_URI']
          if not endpoint:
              database = app.config['PROXY_DATABASE']
              endpoint = f'mysql://{user}:{password}@{host}:{port}/{database}'
  
          self.client = RDSClient(sql_alchemy_url=endpoint, client_kwargs=client_kwargs)
  
          is_the_latest_schema = self.client.validate_schema_version()
          if not is_the_latest_schema:
              LOGGER.warning('Warning: DB Schema is not up to date and it may cause some features not supported. '
                             'Please run rds command to upgrade the schema.')
  
      def is_healthy(self) -> None:
          with self.client.create_session() as session:
              session.execute('SELECT 1 as is_alive')
  
      @timer_with_counter
      def get_user(self, *, id: str) -> Union[User, None]:
          """
          Retrieve user detail based on id(email).
          :param id: the email for the given user
          :return:
          """
          with self.client.create_session() as session:
              user_record = session.query(RDSUser).filter(RDSUser.rk == id).first()
              if not user_record:
                  return user_record
  
              manager_record = user_record.manager
  
              manager_name = ''
              if manager_record and manager_record.full_name:
                  manager_name = manager_record.full_name
  
          return self._build_user_from_record(user_record=user_record, manager_name=manager_name)
  
      @timer_with_counter
      def get_users(self) -> List[User]:
          """
          Retrieve all the user details.
          :return:
          """
          with self.client.create_session() as session:
              users = session.query(RDSUser).filter(RDSUser.is_active.is_(True)).all()
  
          return [self._build_user_from_record(user_record=user) for user in users]
  
      @staticmethod
      def _build_user_from_record(user_record: RDSUser, manager_name: str = '') -> UserEntity:
          return UserEntity(email=user_record.email,
                            first_name=user_record.first_name,
                            last_name=user_record.last_name,
                            full_name=user_record.full_name,
                            is_active=user_record.is_active if user_record.is_active else False,
                            profile_url=user_record.profile_url,
                            github_username=user_record.github_username,
                            team_name=user_record.team_name,
                            slack_id=user_record.slack_id,
                            employee_type=user_record.employee_type,
                            role_name=user_record.role_name,
                            manager_fullname=manager_name)
  
      @timer_with_counter
      def create_update_user(self, *, user: User) -> Tuple[User, bool]:
          """
          Create a user if it does not exist, otherwise update the user.
          :param user:
          :return:
          """
          user_data = UserSchema().dump(user)
          user_record = RDSUser(rk=user.user_id,
                                manager_rk=user.manager_id,
                                published_tag='api_create_update_user',
                                publisher_last_updated_epoch_ms=int(time.time() * 1000))
          for attr, value in user_data.items():
              if hasattr(user_record, attr):
                  user_record.__setattr__(attr, value)
  
          with self.client.create_session() as session:
              existed_user = session.query(RDSUser).filter(RDSUser.rk == user.user_id).first()
              is_new = False if existed_user else True
  
              session.merge(user_record)
              session.commit()
  
          user_result = self._build_user_from_record(user_record)
  
          return user_result, is_new
  
      @timer_with_counter
      def get_table(self, *, table_uri: str) -> Table:
          """
          Retrieve table detail.
          :param table_uri:
          :return:
          """
          with self.client.create_session() as session:
              # table
              table = self._get_table_metadata(session=session, table_uri=table_uri)
              if not table:
                  raise NotFoundException(f'Table URI( {table_uri} ) does not exist')
  
              # columns
              cols = self._get_table_columns(session=session, table_uri=table_uri)
  
              # usage
              readers = self._get_table_readers(session=session, table_uri=table_uri)
  
          table_result = Table(database=table['database'].name,
                               cluster=table['cluster'].name,
                               schema=table['schema'].name,
                               name=table['table'].name,
                               tags=table['tags'],
                               badges=table['badges'],
                               description=table['description'].description if table['description'] else None,
                               columns=cols,
                               owners=table['owners'],
                               table_readers=readers,
                               watermarks=table['watermarks'],
                               table_writer=table['table_writer'],
                               last_updated_timestamp=table['last_updated_timestamp'],
                               source=table['source'],
                               is_view=table['table'].is_view,
                               programmatic_descriptions=table['programmatic_descriptions'])
          return table_result
  
      @timer_with_counter
      def _get_table_metadata(self, *, session: Session, table_uri: str) -> Optional[Dict[str, Any]]:
          table = session.query(RDSTable).filter(RDSTable.rk == table_uri).first()
          if not table:
              return None
  
          schema = table.schema
          cluster = schema.cluster
          database = cluster.database
  
          # watermark
          watermark_results = []
          for watermark in table.watermarks:
              watermark_type = watermark.rk.split('/')[-2]
              watermark_result = Watermark(watermark_type=watermark_type,
                                           partition_key=watermark.partition_key,
                                           partition_value=watermark.partition_value,
                                           create_time=watermark.create_time)
              watermark_results.append(watermark_result)
  
          # tags
          tag_results = []
          tags = [tag for tag in table.tags if tag.tag_type == 'default']
          for tag in tags:
              tag_result = Tag(tag_name=tag.rk, tag_type=tag.tag_type)
              tag_results.append(tag_result)
  
          # badges
          badge_results = []
          for badge in table.badges:
              badge_result = TableBadge(badge_name=badge.rk, category=badge.category)
              badge_results.append(badge_result)
  
          # application
          table_writer = None
          application = table.application
          if application is not None:
              application_id = '' if application.id is None else application.id
              table_writer = Application(application_url=application.application_url,
                                         description=application.description,
                                         name=application.name,
                                         id=application_id)
          # timestamp_value
          timestamp_value = table.timestamp.last_updated_timestamp if table.timestamp else None
  
          # owners
          owner_results = []
          for owner in table.owners:
              owner_results.append(User(email=owner.email))
  
          # source
          source_result = None
          source = table.source
          if source is not None:
              source_result = Source(source_type=source.source_type, source=source.source)
  
          # description
          description_result = table.description
  
          # programmatic descriptions
          prog_description_results = []
          for prog_desc in table.programmatic_descriptions:
              source = prog_desc.description_source
              if source is None:
                  LOGGER.error("A programmatic description with no source was found... skipping.")
              else:
                  prog_description_results.append(ProgrammaticDescription(source=source,
                                                                          text=prog_desc.description))
          prog_description_results.sort(key=lambda x: x.source)
  
          table_metadata_result = dict(database=database,
                                       cluster=cluster,
                                       schema=schema,
                                       table=table,
                                       tags=tag_results,
                                       badges=badge_results,
                                       description=description_result,
                                       owners=owner_results,
                                       watermarks=watermark_results,
                                       table_writer=table_writer,
                                       last_updated_timestamp=timestamp_value,
                                       source=source_result,
                                       programmatic_descriptions=prog_description_results)
  
          return table_metadata_result
  
      @timer_with_counter
      def _get_table_columns(self, *, session: Session, table_uri: str) -> List[Column]:
          # column
          query = session.query(RDSColumn).filter(RDSColumn.table_rk == table_uri)
  
          # description, stats, badges
          query = query.options(
              subqueryload(RDSColumn.description),
              subqueryload(RDSColumn.stats),
              subqueryload(RDSColumn.badges)
          )
  
          columns = query.all()
  
          col_results = []
          for column in columns:
              col_stat_results = []
              for stat in column.stats:
                  col_stat_result = Stat(
                      stat_type=stat.stat_type,
                      stat_val=stat.stat_val,
                      start_epoch=int(float(stat.start_epoch)),
                      end_epoch=int(float(stat.end_epoch))
                  )
                  col_stat_results.append(col_stat_result)
  
              col_badge_results = []
              for badge in column.badges:
                  col_badge_results.append(
                      TableBadge(badge_name=badge.rk, category=badge.category)
                  )
  
              col_result = Column(name=column.name,
                                  description=column.description.description
                                  if column.description else None,
                                  col_type=column.type,
                                  sort_order=int(column.sort_order),
                                  stats=col_stat_results,
                                  badges=col_badge_results)
              col_results.append(col_result)
  
          return col_results
  
      @timer_with_counter
      def _get_table_readers(self, *, session: Session, table_uri: str) -> List[Reader]:
          readers = session.query(RDSTableUsage).filter(
              RDSTableUsage.table_rk == table_uri
          ).order_by(RDSTableUsage.read_count).limit(5).all()
  
          reader_results = []
          for reader in readers:
              reader_result = Reader(user=User(email=reader.user_rk),
                                     read_count=reader.read_count)
              reader_results.append(reader_result)
  
          return reader_results
  
      @timer_with_counter
      def delete_owner(self, *, table_uri: str, owner: str) -> None:
          """
          Delete relation between the given table and owner.
          :param table_uri:
          :param owner:
          :return:
          """
          try:
              with self.client.create_session() as session:
                  session.query(RDSTableOwner).filter(
                      RDSTableOwner.table_rk == table_uri,
                      RDSTableOwner.user_rk == owner
                  ).delete()
                  session.commit()
          except Exception as e:
              LOGGER.exception(f'Failed to delete owner {owner} for table {table_uri}')
              raise e
  
      @timer_with_counter
      def add_owner(self, *, table_uri: str, owner: str) -> None:
          """
          Add the owner for the given table.
          :param table_uri:
          :param owner:
          :return:
          """
          user = RDSUser(rk=owner, email=owner)
          table_owner = RDSTableOwner(table_rk=table_uri, user_rk=owner)
          try:
              with self.client.create_session() as session:
                  session.merge(user)
                  session.merge(table_owner)
                  session.commit()
          except Exception as e:
              LOGGER.exception(f'Failed to add owner {owner} for table {table_uri}')
              raise e
  
      @timer_with_counter
      def get_table_description(self, *, table_uri: str) -> Union[str, None]:
          """
          Get the table description based on table uri.
          :param table_uri:
          :return:
          """
          with self.client.create_session() as session:
              description = session.query(RDSTableDescription.description).filter(
                  RDSTableDescription.table_rk == table_uri
              ).scalar()
  
          return description
  
      @timer_with_counter
      def put_table_description(self, *, table_uri: str, description: str) -> None:
          """
          Update table description with one from user
          :param table_uri:
          :param description:
          :return:
          """
          desc_key = table_uri + '/_description'
          description = RDSTableDescription(rk=desc_key,
                                            description_source='description',
                                            description=description,
                                            table_rk=table_uri)
          try:
              with self.client.create_session() as session:
                  session.merge(description)
                  session.commit()
          except Exception as e:
              LOGGER.exception(f'Failed to add description for table {table_uri}')
              raise e
  
      @timer_with_counter
      def add_tag(self, *,
                  id: str,
                  tag: str,
                  tag_type: str = 'default',
                  resource_type: ResourceType = ResourceType.Table) -> None:
          """
          Add a new tag if it does not exist and add the relation between the given resource and tag.
          :param id:
          :param tag:
          :param tag_type:
          :param resource_type:
          :return:
          """
          LOGGER.info(f'New tag {tag} for id {id} with type {tag_type} and resource type {resource_type.name}')
  
          resource_table = f'{resource_type.name.lower()}_tag'
          resource_model = self._get_model_from_table_name(resource_table)
          if not resource_model:
              raise NotImplementedError(f'The resource type {resource_type.name} is not defined!')
  
          resource_key = f'{resource_type.name.lower()}_rk'
  
          tag_record = RDSTag(rk=tag, tag_type=tag_type)
          resource_tag_record = resource_model(tag_rk=tag)
          resource_tag_record.__setattr__(resource_key, id)
          try:
              with self.client.create_session() as session:
                  session.merge(tag_record)
                  session.merge(resource_tag_record)
                  session.commit()
          except Exception as e:
              LOGGER.exception(f'Failed to add tag {tag} for {id}')
              raise e
  
      @timer_with_counter
      def add_badge(self, *,
                    id: str,
                    badge_name: str,
                    category: str = '',
                    resource_type: ResourceType = ResourceType.Table) -> None:
          """
          Add a new badge if it does not exist and add the relation between the given resource and badge.
          :param id:
          :param badge_name:
          :param category:
          :param resource_type:
          :return:
          """
          LOGGER.info(f'New badge {badge_name} for id {id} with category {category} '
                      f'and resource type {resource_type.name}')
  
          resource_table = f'{resource_type.name.lower()}_badge'
          resource_model = self._get_model_from_table_name(resource_table)
          if not resource_model:
              raise NotImplementedError(f'The resource type {resource_type.name} is not defined!')
  
          resource_key = f'{resource_type.name.lower()}_rk'
  
          badge_record = RDSBadge(rk=badge_name, category=category)
          resource_badge_record = resource_model(badge_rk=badge_name)
          resource_badge_record.__setattr__(resource_key, id)
          try:
              with self.client.create_session() as session:
                  session.merge(badge_record)
                  session.merge(resource_badge_record)
                  session.commit()
          except Exception as e:
              LOGGER.exception(f'Failed to add badge {badge_name} for {id}')
              raise e
  
      @timer_with_counter
      def delete_tag(self, *,
                     id: str,
                     tag: str,
                     tag_type: str = 'default',
                     resource_type: ResourceType = ResourceType.Table) -> None:
          """
          Delete the relation between the resource and the tag.
          :param id:
          :param tag:
          :param tag_type:
          :param resource_type:
          :return:
          """
          LOGGER.info(f'Delete tag {tag} for {id} with type {tag_type} and resource_type: {resource_type.name}')
  
          resource_table = f'{resource_type.name.lower()}_tag'
          resource_model = self._get_model_from_table_name(resource_table)
          if not resource_model:
              raise NotImplementedError(f'{resource_type.name} is not defined!')
  
          resource_key = f'{resource_type.name.lower()}_rk'
          resource_attr = getattr(resource_model, resource_key)
          tag_attr = getattr(resource_model, 'tag_rk')
          try:
              with self.client.create_session() as session:
                  session.query(resource_model).filter(resource_attr == id, tag_attr == tag).delete()
                  session.commit()
          except Exception as e:
              LOGGER.exception(f'Failed to delete tag {tag} for {id}')
              raise e
  
      @timer_with_counter
      def delete_badge(self, *,
                       id: str,
                       badge_name: str,
                       category: str,
                       resource_type: ResourceType = ResourceType.Table) -> None:
          """
          Delete the relation between the resource and the badge.
          :param id:
          :param badge_name:
          :param category:
          :param resource_type:
          :return:
          """
          LOGGER.info(f'Delete badge {badge_name} for {id} with {category}')
  
          resource_table = f'{resource_type.name.lower()}_badge'
          resource_model = self._get_model_from_table_name(resource_table)
          if not resource_model:
              raise NotImplementedError(f'{resource_type.name} is not defined!')
  
          resource_key = f'{resource_type.name.lower()}_rk'
          resource_attr = getattr(resource_model, resource_key)
          badge_attr = getattr(resource_model, 'badge_rk')
          try:
              with self.client.create_session() as session:
                  session.query(resource_model).filter(resource_attr == id, badge_attr == badge_name).delete()
                  session.commit()
          except Exception as e:
              LOGGER.exception(f'Failed to delete badge {badge_name} for {id}')
              raise e
  
      @staticmethod
      def _get_model_from_table_name(table_name: str) -> Optional[Type[RDSModel]]:
          """
          Get rds model for the given table name.
          :param table_name:
          :return:
          """
          table_model = None
          try:
              if hasattr(Base, '_decl_class_registry'):
                  models = Base._decl_class_registry.values()  # sqlalchemy < 1.4
              else:
                  models = Base.registry._class_registry.values()
  
              for model in models:
                  if hasattr(model, '__tablename__') and model.__tablename__ == table_name:
                      table_model = model
          except Exception as e:
              LOGGER.exception(f'Failed to get model for the table: {table_name} from rds model base')
              raise e
  
          return table_model
  
      @timer_with_counter
      def put_column_description(self, *, table_uri: str, column_name: str, description: str) -> None:
          """
          Update column description with input from user.
          :param table_uri:
          :param column_name:
          :param description:
          :return:
          """
          column_uri = table_uri + '/' + column_name
          desc_key = column_uri + '/_description'
          description = RDSColumnDescription(rk=desc_key,
                                             description_source='description',
                                             description=description,
                                             column_rk=column_uri)
          try:
              with self.client.create_session() as session:
                  session.merge(description)
                  session.commit()
          except Exception as e:
              LOGGER.exception(f'Failed to update the table {table_uri} column {column_name} description')
              raise e
  
      @timer_with_counter
      def get_column_description(self, *, table_uri: str, column_name: str) -> Union[str, None]:
          """
          Get the column description based on table uri.
          :param table_uri:
          :param column_name:
          :return:
          """
          column_uri = table_uri + '/' + column_name
          desc_key = column_uri + '/_description'
          with self.client.create_session() as session:
              description = session.query(RDSColumnDescription.description).filter(
                  RDSColumnDescription.rk == desc_key
              ).scalar()
  
          return description
  
      @timer_with_counter
      def get_popular_tables(self, *, num_entries: int, user_id: Optional[str] = None) -> List[PopularTable]:
          """
          Retrieve popular tables. As popular table computation requires full scan of table usage,
          it will utilize cached method _get_popular_tables_uris.
          :param num_entries:
          :param user_id:
          :return:
          """
          if user_id is None:
              table_uris = self._get_global_popular_resources_uris(num_entries=num_entries)
          else:
              table_uris = self._get_personal_popular_resources_uris(num_entries=num_entries, user_id=user_id)
  
          if not table_uris:
              return []
  
          with self.client.create_session() as session:
              # table
              query = session.query(RDSTable).filter(RDSTable.rk.in_(table_uris))
  
              # description
              query = query.options(
                  subqueryload(RDSTable.description).options(
                      load_only(RDSTableDescription.description)
                  )
              )
  
              # schema, cluster, database
              query = query.options(
                  subqueryload(RDSTable.schema).options(
                      load_only(RDSSchema.name, RDSSchema.cluster_rk),
                      subqueryload(RDSSchema.cluster).options(
                          load_only(RDSCluster.name, RDSCluster.database_rk),
                          subqueryload(RDSCluster.database).options(
                              load_only(RDSDatabase.name)
                          )
                      )
                  )
              )
  
              tables = query.all()
  
          popular_tables = []
          for table in tables:
              schema = table.schema
              cluster = schema.cluster
              database = cluster.database
              description = table.description
              popular_table = PopularTable(database=database.name,
                                           cluster=cluster.name,
                                           schema=schema.name,
                                           name=table.name,
                                           description=description.description if description else None)
              popular_tables.append(popular_table)
  
          return popular_tables
  
      @timer_with_counter
      def get_popular_resources(self, *,
                                num_entries: int,
                                resource_types: List[str],
                                user_id: Optional[str] = None) -> Dict[str, List]:
          """
          Retrieve popular resources. As popular resource computation requires full scan of resource usage,
          it will cached popular resources uris.
          :param num_entries:
          :param resource_types:
          :param user_id:
          :return:
          """
          popular_resources: Dict[str, List] = dict()
          for resource in resource_types:
              resource_type = to_resource_type(label=resource)
              popular_resources[resource_type.name] = list()
              if user_id is None:
                  # Get global popular Table/Dashboard URIs
                  resource_uris = self._get_global_popular_resources_uris(num_entries=num_entries,
                                                                          resource_type=resource_type)
              else:
                  # Get personalized popular Table/Dashboard URIs
                  resource_uris = self._get_personal_popular_resources_uris(num_entries=num_entries,
                                                                            user_id=user_id,
                                                                            resource_type=resource_type)
  
              if resource_type == ResourceType.Table:
                  popular_resources[resource_type.name] = self._get_popular_tables(table_uris=resource_uris)
              elif resource_type == ResourceType.Dashboard:
                  popular_resources[resource_type.name] = self._get_popular_dashboards(dashboard_uris=resource_uris)
  
          return popular_resources
  
      @_CACHE.cache('_get_global_popular_resources_uris', expire=_GET_POPULAR_RESOURCES_CACHE_EXPIRY_SEC)
      def _get_global_popular_resources_uris(self,
                                             num_entries: int,
                                             resource_type: ResourceType = ResourceType.Table) -> List[str]:
          """
          Retrieve popular resources uris. Will provide resources with top x popularity score.
          Popularity score = number of distinct readers * log(total number of reads)
          The result of this method will be cached based on the key (num_entries),
          and the cache will be expired based on _GET_POPULAR_RESOURCES_CACHE_EXPIRY_SEC
          :param num_entries:
          :param resource_type:
          :return:
          """
          LOGGER.info('Querying global popular resources URIs')
  
          num_readers = app.config['POPULAR_RESOURCES_MINIMUM_READER_COUNT']
  
          relation_model = resource_relation_model[resource_type][UserResourceRel.read]
          res_key = f'{resource_type.name.lower()}_rk'
          res_attr = getattr(relation_model, res_key)
          user_attr = getattr(relation_model, 'user_rk')
          read_count_attr = getattr(relation_model, 'read_count')
  
          with self.client.create_session() as session:
              readers = func.count(user_attr).label('readers')
              usage_subquery = session.query(
                  res_attr.label('res_key'),
                  readers,
                  func.sum(read_count_attr).label('total_reads')
              ).group_by(res_attr).having(readers >= num_readers).subquery()
  
  