content: # Copyright Contributors to the Amundsen project.
  # SPDX-License-Identifier: Apache-2.0
  
  import logging
  import re
  import textwrap
  import time
  from random import randint
  from typing import (Any, Dict, Iterable, List, Optional, Tuple,  # noqa: F401
                      Union, no_type_check)
  
  import neo4j
  from amundsen_common.entity.resource_type import ResourceType, to_resource_type
  from amundsen_common.models.api import health_check
  from amundsen_common.models.dashboard import DashboardSummary
  from amundsen_common.models.feature import Feature, FeatureWatermark
  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, Source, SqlJoin,
                                            SqlWhere, Stat, Table, TableSummary,
                                            Tag, TypeMetadata, User, Watermark)
  from amundsen_common.models.user import User as UserEntity
  from amundsen_common.models.user import UserSchema
  from beaker.cache import CacheManager
  from beaker.util import parse_cache_config_options
  from flask import current_app, has_app_context
  from neo4j import GraphDatabase, Record, Transaction  # noqa: F401
  from neo4j.api import (SECURITY_TYPE_SECURE,
                         SECURITY_TYPE_SELF_SIGNED_CERTIFICATE, parse_neo4j_uri)
  from neo4j.exceptions import ClientError
  
  from metadata_service import config
  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)
  
  CREATED_EPOCH_MS = 'publisher_created_epoch_ms'
  LAST_UPDATED_EPOCH_MS = 'publisher_last_updated_epoch_ms'
  PUBLISHED_TAG_PROPERTY_NAME = 'published_tag'
  
  LOGGER = logging.getLogger(__name__)
  
  
  def execute_statement(tx: Transaction, stmt: str, params: Optional[dict] = None) -> List[Record]:
      """
      Executes statement against Neo4j. If execution fails, it rollsback and raises exception.
      """
      LOGGER.debug('Executing statement: %s with params %s', stmt, params)
  
      result = tx.run(stmt, parameters=params)
      return [record for record in result]
  
  
  def get_single_record(records_list: List[Record]) -> Record:
      """
      Helper method to get single item from _execute_cypher_query return when only one item is expected.
      Emulates neo4j's Result.single() behavior.
      """
      records_list_length = len(records_list)
      if records_list_length > 1:
          LOGGER.warning(f'There are {records_list_length} records in this result but only 1 was expected')
      try:
          return records_list[0]
      except IndexError as e:
          return None
  
  
  class Neo4jProxy(BaseProxy):
      """
      A proxy to Neo4j (Gateway to Neo4j)
      """
  
      def __init__(self, *,
                   host: str,
                   port: int = 7687,
                   user: str = 'neo4j',
                   password: str = '',
                   num_conns: int = 50,
                   max_connection_lifetime_sec: int = 100,
                   encrypted: bool = False,
                   validate_ssl: bool = False,
                   database_name: str = neo4j.DEFAULT_DATABASE,
                   **kwargs: dict) -> None:
          """
          There's currently no request timeout from client side where server
          side can be enforced via "dbms.transaction.timeout"
          By default, it will set max number of connections to 50 and connection time out to 10 seconds.
          :param endpoint: neo4j endpoint
          :param num_conns: number of connections
          :param max_connection_lifetime_sec: max lifetime the connection can have when it comes to reuse. In other
          words, connection lifetime longer than this value won't be reused and closed on garbage collection. This
          value needs to be smaller than surrounding network environment's timeout.
          :param database_name: the neo4j database to be queried if different from the default
          """
          endpoint = f'{host}:{port}'
  
          self._database_name = database_name
  
          driver_args = {
              'uri': endpoint,
              'max_connection_lifetime': max_connection_lifetime_sec,
              'auth': (user, password),
              'connection_timeout': 10,
              'max_connection_pool_size': num_conns,
          }
  
          # if URI scheme not secure set `trust`` and `encrypted` arguments for the driver
          # https://neo4j.com/docs/api/python-driver/current/api.html#uri
          _, security_type, _ = parse_neo4j_uri(uri=endpoint)
          if security_type not in [SECURITY_TYPE_SELF_SIGNED_CERTIFICATE, SECURITY_TYPE_SECURE]:
              trust = neo4j.TRUST_SYSTEM_CA_SIGNED_CERTIFICATES if validate_ssl else neo4j.TRUST_ALL_CERTIFICATES
              default_security_conf = {'trust': trust, 'encrypted': encrypted}
              driver_args.update(default_security_conf)
  
          self._driver = GraphDatabase.driver(**driver_args)
  
      def health(self) -> health_check.HealthCheck:
          """
          Runs one or more series of checks on the service. Can also
          optionally return additional metadata about each check (e.g.
          latency to database, cpu utilization, etc.).
          """
          checks = {}
          try:
              # dbms.cluster.overview() is only available for enterprise neo4j users
              cluster_overview = self._execute_cypher_query(statement='CALL dbms.cluster.overview()', param_dict={})
              checks = dict(cluster_overview[0])
              checks['overview_enabled'] = True
              status = health_check.OK
          except ClientError:
              checks = {'overview_enabled': False}
              status = health_check.OK  # Can connect to database but plugin is not available
          except Exception:
              status = health_check.FAIL
          final_checks = {f'{type(self).__name__}:connection': checks}
          return health_check.HealthCheck(status=status, checks=final_checks)
  
      @timer_with_counter
      def get_table(self, *, table_uri: str) -> Table:
          """
          :param table_uri: Table URI
          :return:  A Table object
          """
          cols, last_neo4j_record = self._exec_col_query(table_uri)
  
          readers = self._exec_usage_query(table_uri)
          owners = self._exec_owners_query(table_uri)
  
          wmk_results, table_writer, table_apps, timestamp_value, tags, source, \
              badges, prog_descs, resource_reports = self._exec_table_query(table_uri)
  
          joins, filters = self._exec_table_query_query(table_uri)
  
          table = Table(database=last_neo4j_record['db']['name'],
                        cluster=last_neo4j_record['clstr']['name'],
                        schema=last_neo4j_record['schema']['name'],
                        name=last_neo4j_record['tbl']['name'],
                        tags=tags,
                        badges=badges,
                        description=self._safe_get(last_neo4j_record, 'tbl_dscrpt', 'description'),
                        columns=cols,
                        owners=owners,
                        table_readers=readers,
                        watermarks=wmk_results,
                        table_writer=table_writer,
                        table_apps=table_apps,
                        last_updated_timestamp=timestamp_value,
                        source=source,
                        is_view=self._safe_get(last_neo4j_record, 'tbl', 'is_view'),
                        programmatic_descriptions=prog_descs,
                        common_joins=joins,
                        common_filters=filters,
                        resource_reports=resource_reports
                        )
  
          return table
  
      @timer_with_counter
      def _exec_col_query(self, table_uri: str) -> Tuple:
          # Return Value: (Columns, Last Processed Record)
  
          column_level_query = textwrap.dedent("""
          MATCH (db:Database)-[:CLUSTER]->(clstr:Cluster)-[:SCHEMA]->(schema:Schema)
          -[:TABLE]->(tbl:Table {key: $tbl_key})-[:COLUMN]->(col:Column)
          OPTIONAL MATCH (tbl)-[:DESCRIPTION]->(tbl_dscrpt:Description)
          OPTIONAL MATCH (col:Column)-[:DESCRIPTION]->(col_dscrpt:Description)
          OPTIONAL MATCH (col:Column)-[:STAT]->(stat:Stat)
          OPTIONAL MATCH (col:Column)-[:HAS_BADGE]->(badge:Badge)
          OPTIONAL MATCH (col:Column)-[:TYPE_METADATA]->(Type_Metadata)-[:SUBTYPE *0..]->(tm:Type_Metadata)
          OPTIONAL MATCH (tm:Type_Metadata)-[:DESCRIPTION]->(tm_dscrpt:Description)
          OPTIONAL MATCH (tm:Type_Metadata)-[:HAS_BADGE]->(tm_badge:Badge)
          WITH db, clstr, schema, tbl, tbl_dscrpt, col, col_dscrpt, collect(distinct stat) as col_stats,
          collect(distinct badge) as col_badges,
          {node: tm, description: tm_dscrpt, badges: collect(distinct tm_badge)} as tm_results
          RETURN db, clstr, schema, tbl, tbl_dscrpt, col, col_dscrpt, col_stats, col_badges,
          collect(distinct tm_results) as col_type_metadata
          ORDER BY col.sort_order;""")
  
          tbl_col_neo4j_records = self._execute_cypher_query(
              statement=column_level_query, param_dict={'tbl_key': table_uri})
  
          cols = []
          last_neo4j_record = None
          for tbl_col_neo4j_record in tbl_col_neo4j_records:
              # Getting last record from this for loop as Neo4j's result's random access is O(n) operation.
              col_stats = []
              for stat in tbl_col_neo4j_record['col_stats']:
                  col_stat = 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_stats.append(col_stat)
  
              column_badges = self._make_badges(tbl_col_neo4j_record['col_badges'])
  
              col_type_metadata = self._get_type_metadata(tbl_col_neo4j_record['col_type_metadata'])
  
              last_neo4j_record = tbl_col_neo4j_record
              col = Column(name=tbl_col_neo4j_record['col']['name'],
                           description=self._safe_get(tbl_col_neo4j_record, 'col_dscrpt', 'description'),
                           col_type=tbl_col_neo4j_record['col']['col_type'],
                           sort_order=int(tbl_col_neo4j_record['col']['sort_order']),
                           stats=col_stats,
                           badges=column_badges,
                           type_metadata=col_type_metadata)
  
              cols.append(col)
  
          if not cols:
              raise NotFoundException('Table URI( {table_uri} ) does not exist'.format(table_uri=table_uri))
  
          return sorted(cols, key=lambda item: item.sort_order), last_neo4j_record
  
      def _get_type_metadata(self, type_metadata_results: List) -> Optional[TypeMetadata]:
          """
          Generates a TypeMetadata object for a column. All columns will have at least
          one associated type metadata node if the ComplexTypeTransformer is configured
          to transform table metadata. Otherwise, there will be no type metadata found
          and this will return quickly.
  
          :param type_metadata_results: A list of type metadata values for a column
          :return: a TypeMetadata object
          """
          # If there are no Type_Metadata nodes, type_metadata_results will have
          # one object with an empty node value
          if len(type_metadata_results) > 0 and type_metadata_results[0]['node'] is not None:
              sorted_type_metadata = sorted(type_metadata_results, key=lambda x: x['node']['key'])
          else:
              return None
  
          type_metadata_nodes: Dict[str, TypeMetadata] = {}
          type_metadata_children: Dict[str, Dict] = {}
          for tm in sorted_type_metadata:
              tm_node = tm['node']
              description = self._safe_get(tm, 'description', 'description')
              sort_order = self._safe_get(tm_node, 'sort_order') or 0
              badges = self._safe_get(tm, 'badges')
              # kind refers to the general type of the TypeMetadata, such as "array" or "map",
              # while data_type refers to the entire type such as "array<int>" or "map<string, string>"
              type_metadata = TypeMetadata(kind=tm_node['kind'], name=tm_node['name'], key=tm_node['key'],
                                           description=description, data_type=tm_node['data_type'],
                                           sort_order=sort_order, badges=self._make_badges(badges) if badges else [])
  
              # type_metadata_nodes maps each type metadata path to its corresponding TypeMetadata object
              tm_key_regex = re.compile(
                  r'(?P<db>\w+):\/\/(?P<cluster>\w+)\.(?P<schema>\w+)\/(?P<tbl>\w+)\/(?P<col>\w+)\/type\/(?P<tm_path>.*)'
              )
              tm_key_match = tm_key_regex.search(type_metadata.key)
              if tm_key_match is None:
                  LOGGER.error(f'Could not retrieve the type metadata path from key {type_metadata.key}')
                  continue
              tm_path = tm_key_match.group('tm_path')
              type_metadata_nodes[tm_path] = type_metadata
  
              # type_metadata_children is a nested dict where each type metadata node name
              # maps to a dict of its children's names
              split_key_list = tm_path.split('/')
              tm_name = split_key_list.pop()
              node_children = self._safe_get(type_metadata_children, *split_key_list)
              if node_children is not None:
                  node_children[tm_name] = {}
              else:
                  LOGGER.error(f'Could not construct the dict of children for type metadata key {type_metadata.key}')
  
          # Iterate over the temporary children dict to create the proper TypeMetadata structure
          result = self._build_type_metadata_structure('', type_metadata_children, type_metadata_nodes)
          return result[0] if len(result) > 0 else None
  
      def _build_type_metadata_structure(self, prev_path: str, tm_children: Dict, tm_nodes: Dict) -> List[TypeMetadata]:
          type_metadata = []
          for node_name, children in tm_children.items():
              curr_path = f'{prev_path}/{node_name}' if prev_path else node_name
              tm = tm_nodes.get(curr_path)
              if tm is None:
                  LOGGER.error(f'Could not find expected type metadata object at type metadata path {curr_path}')
                  continue
              if len(children) > 0:
                  tm.children = self._build_type_metadata_structure(curr_path, children, tm_nodes)
              type_metadata.append(tm)
  
          if len(type_metadata) > 1:
              type_metadata.sort(key=lambda x: x.sort_order)
  
          return type_metadata
  
      @timer_with_counter
      def _exec_usage_query(self, table_uri: str) -> List[Reader]:
          # Return Value: List[Reader]
  
          usage_query = textwrap.dedent("""\
          MATCH (user:User)-[read:READ]->(table:Table {key: $tbl_key})
          RETURN user.email as email, read.read_count as read_count, table.name as table_name
          ORDER BY read.read_count DESC LIMIT 5;
          """)
  
          usage_neo4j_records = self._execute_cypher_query(statement=usage_query,
                                                           param_dict={'tbl_key': table_uri})
          readers = []  # type: List[Reader]
          for usage_neo4j_record in usage_neo4j_records:
              reader_data = self._get_user_details(user_id=usage_neo4j_record['email'])
              reader = Reader(user=self._build_user_from_record(record=reader_data),
                              read_count=usage_neo4j_record['read_count'])
              readers.append(reader)
  
          return readers
  
      @timer_with_counter
      def _exec_owners_query(self, table_uri: str) -> List[User]:
          # Return Value: List[User]
          owners_query = textwrap.dedent("""
              MATCH (owner:User)<-[:OWNER]-(tbl:Table {key: $tbl_key})
              RETURN collect(distinct owner) as owner_records
          """)
          owners_neo4j_records = self._execute_cypher_query(statement=owners_query,
                                                            param_dict={'tbl_key': table_uri})
  
          owners_neo4j_records = get_single_record(owners_neo4j_records)
  
          owners = []  # type: List[User]
          for owner_neo4j_record in owners_neo4j_records.get('owner_records', []):
              owner_data = self._get_user_details(user_id=owner_neo4j_record['email'])
              owner = self._build_user_from_record(record=owner_data)
              owners.append(owner)
  
          return owners
  
      @timer_with_counter
      def _exec_table_query(self, table_uri: str) -> Tuple:
          """
          Queries one Cypher record with watermark list, Application,
          ,timestamp, and tag records.
          """
  
          # Return Value: (Watermark Results, Table Writer, Last Updated Timestamp, tag records)
  
          table_level_query = textwrap.dedent("""\
          MATCH (tbl:Table {key: $tbl_key})
          OPTIONAL MATCH (wmk:Watermark)-[:BELONG_TO_TABLE]->(tbl)
          OPTIONAL MATCH (app_producer:Application)-[:GENERATES]->(tbl)
          OPTIONAL MATCH (app_consumer:Application)-[:CONSUMES]->(tbl)
          OPTIONAL MATCH (tbl)-[:LAST_UPDATED_AT]->(t:Timestamp)
          OPTIONAL MATCH (tbl)-[:TAGGED_BY]->(tag:Tag{tag_type: $tag_normal_type})
          OPTIONAL MATCH (tbl)-[:HAS_BADGE]->(badge:Badge)
          OPTIONAL MATCH (tbl)-[:SOURCE]->(src:Source)
          OPTIONAL MATCH (tbl)-[:DESCRIPTION]->(prog_descriptions:Programmatic_Description)
          OPTIONAL MATCH (tbl)-[:HAS_REPORT]->(resource_reports:Report)
          RETURN collect(distinct wmk) as wmk_records,
          collect(distinct app_producer) as producing_apps,
          collect(distinct app_consumer) as consuming_apps,
          t.last_updated_timestamp as last_updated_timestamp,
          collect(distinct tag) as tag_records,
          collect(distinct badge) as badge_records,
          src,
          collect(distinct prog_descriptions) as prog_descriptions,
          collect(distinct resource_reports) as resource_reports
          """)
  
          table_records = self._execute_cypher_query(statement=table_level_query,
                                                     param_dict={'tbl_key': table_uri,
                                                                 'tag_normal_type': 'default'})
  
          table_records = get_single_record(table_records)
  
          wmk_results = []
          wmk_records = table_records['wmk_records']
          for record in wmk_records:
              if record['key'] is not None:
                  watermark_type = record['key'].split('/')[-2]
                  wmk_result = Watermark(watermark_type=watermark_type,
                                         partition_key=record['partition_key'],
                                         partition_value=record['partition_value'],
                                         create_time=record['create_time'])
                  wmk_results.append(wmk_result)
  
          tags = []
          if table_records.get('tag_records'):
              tag_records = table_records['tag_records']
              for record in tag_records:
                  tag_result = Tag(tag_name=record['key'],
                                   tag_type=record['tag_type'])
                  tags.append(tag_result)
  
          # this is for any badges added with BadgeAPI instead of TagAPI
          badges = self._make_badges(table_records.get('badge_records'))
  
          table_writer, table_apps = self._create_apps(table_records['producing_apps'], table_records['consuming_apps'])
  
          timestamp_value = table_records['last_updated_timestamp']
  
          src = None
  
          if table_records['src']:
              src = Source(source_type=table_records['src']['source_type'],
                           source=table_records['src']['source'])
  
          prog_descriptions = self._extract_programmatic_descriptions_from_query(
              table_records.get('prog_descriptions', [])
          )
  
          resource_reports = self._extract_resource_reports_from_query(table_records.get('resource_reports', []))
  
          return wmk_results, table_writer, table_apps, timestamp_value, \
              tags, src, badges, prog_descriptions, resource_reports
  
      @timer_with_counter
      def _exec_table_query_query(self, table_uri: str) -> Tuple:
          """
          Queries one Cypher record with results that contain information about queries
          and entities (e.g. joins, where clauses, etc.) associated to queries that are executed
          on the table.
          """
  
          # Return Value: (Watermark Results, Table Writer, Last Updated Timestamp, tag records)
          table_query_level_query = textwrap.dedent("""
          MATCH (tbl:Table {key: $tbl_key})
          OPTIONAL MATCH (tbl)-[:COLUMN]->(col:Column)-[COLUMN_JOINS_WITH]->(j:Join)
          OPTIONAL MATCH (j)-[JOIN_OF_COLUMN]->(col2:Column)
          OPTIONAL MATCH (j)-[JOIN_OF_QUERY]->(jq:Query)-[:HAS_EXECUTION]->(exec:Execution)
          WITH tbl, j, col, col2,
              sum(coalesce(exec.execution_count, 0)) as join_exec_cnt
          ORDER BY join_exec_cnt desc
          LIMIT 5
          WITH tbl,
              COLLECT(DISTINCT {
              join: {
                  joined_on_table: {
                      database: case when j.left_table_key = $tbl_key
                                then j.right_database
                                else j.left_database
                                end,
                      cluster: case when j.left_table_key = $tbl_key
                               then j.right_cluster
                               else j.left_cluster
                               end,
                      schema: case when j.left_table_key = $tbl_key
                              then j.right_schema
                              else j.left_schema
                              end,
                      name: case when j.left_table_key = $tbl_key
                            then j.right_table
                            else j.left_table
                            end
                  },
                  joined_on_column: col2.name,
                  column: col.name,
                  join_type: j.join_type,
                  join_sql: j.join_sql
              },
              join_exec_cnt: join_exec_cnt
          }) as joins
          WITH tbl, joins
          OPTIONAL MATCH (tbl)-[:COLUMN]->(col:Column)-[USES_WHERE_CLAUSE]->(whr:Where)
          OPTIONAL MATCH (whr)-[WHERE_CLAUSE_OF]->(wq:Query)-[:HAS_EXECUTION]->(whrexec:Execution)
          WITH tbl, joins,
              whr, sum(coalesce(whrexec.execution_count, 0)) as where_exec_cnt
          ORDER BY where_exec_cnt desc
          LIMIT 5
          RETURN tbl, joins,
            COLLECT(DISTINCT {
              where_clause: whr.where_clause,
              where_exec_cnt: where_exec_cnt
            }) as filters
          """)
  
          query_records = self._execute_cypher_query(statement=table_query_level_query, param_dict={'tbl_key': table_uri})
  
          table_query_records = get_single_record(query_records)
  
          joins = self._extract_joins_from_query(table_query_records.get('joins', [{}]))
          filters = self._extract_filters_from_query(table_query_records.get('filters', [{}]))
  
          return joins, filters
  
      def _extract_programmatic_descriptions_from_query(self, raw_prog_descriptions: dict) -> list:
          prog_descriptions = []
          for prog_description in raw_prog_descriptions:
              source = prog_description['description_source']
              if source is None:
                  LOGGER.error("A programmatic description with no source was found... skipping.")
              else:
                  prog_descriptions.append(ProgrammaticDescription(source=source, text=prog_description['description']))
          prog_descriptions.sort(key=lambda x: x.source)
          return prog_descriptions
  
      def _extract_resource_reports_from_query(self, raw_resource_reports: dict) -> list:
          resource_reports = []
          for resource_report in raw_resource_reports:
              name = resource_report.get('name')
              if name is None:
                  LOGGER.error("A report with no name found... skipping.")
              else:
                  resource_reports.append(ResourceReport(name=name, url=resource_report['url']))
  
          parsed_reports = current_app.config['RESOURCE_REPORT_CLIENT'](resource_reports) \
              if current_app.config['RESOURCE_REPORT_CLIENT'] else resource_reports
  
          parsed_reports.sort(key=lambda x: x.name)
  
          return parsed_reports
  
      def _extract_joins_from_query(self, joins: List[Dict]) -> List[SqlJoin]:
          valid_joins = []
          for join in joins:
              join_data = join['join']
              if all(join_data.values()):
                  new_sql_join = SqlJoin(join_sql=join_data['join_sql'],
                                         join_type=join_data['join_type'],
                                         joined_on_column=join_data['joined_on_column'],
                                         joined_on_table=TableSummary(**join_data['joined_on_table']),
                                         column=join_data['column'])
                  valid_joins.append(new_sql_join)
          return valid_joins
  
      def _extract_filters_from_query(self, filters: List[Dict]) -> List[SqlWhere]:
          return_filters = []
          for filt in filters:
              filter_where = filt.get('where_clause')
              if filter_where:
                  return_filters.append(SqlWhere(where_clause=filter_where))
          return return_filters
  
      @no_type_check
      def _safe_get(self, dct, *keys):
          """
          Helper method for getting value from nested dict. This also works either key does not exist or value is None.
          :param dct:
          :param keys:
          :return:
          """
          for key in keys:
              dct = dct.get(key)
              if dct is None:
                  return None
          return dct
  
      @timer_with_counter
      def _execute_cypher_query(self, *,
                                statement: str,
                                param_dict: Dict[str, Any]) -> List[Record]:
          """
          Execute Cypher queries using managed read transactions
          """
          if LOGGER.isEnabledFor(logging.DEBUG):
              LOGGER.debug('Executing Cypher query: {statement} with params {params}: '.format(statement=statement,
                                                                                               params=param_dict))
          start = time.time()
          try:
              with self._driver.session(database=self._database_name) as session:
                  return session.read_transaction(execute_statement, statement, param_dict)
  
          finally:
              # TODO: Add support on statsd
              if LOGGER.isEnabledFor(logging.DEBUG):
                  LOGGER.debug('Cypher query execution elapsed for {} seconds'.format(time.time() - start))
  
      # noinspection PyMethodMayBeStatic
      def _make_badges(self, badges: Iterable) -> List[Badge]:
          """
          Generates a list of Badges objects
  
          :param badges: A list of badges of a table, column, or type_metadata
          :return: a list of Badge objects
          """
          _badges = []
          for badge in badges:
              _badges.append(Badge(badge_name=badge["key"], category=badge["category"]))
          return _badges
  
      @timer_with_counter
      def get_resource_description(self, *,
                                   resource_type: ResourceType,
                                   uri: str) -> Description:
          """
          Get the resource description based on the uri. Any exception will propagate back to api server.
  
          :param resource_type:
          :param id:
          :return:
          """
  
          description_query = textwrap.dedent("""
          MATCH (n:{node_label} {{key: $key}})-[:DESCRIPTION]->(d:Description)
          RETURN d.description AS description;
          """.format(node_label=resource_type.name))
  
          result = self._execute_cypher_query(statement=description_query,
                                              param_dict={'key': uri})
  
          result = get_single_record(result)
          return Description(description=result['description'] if result else None)  # type: ignore
  
      @timer_with_counter
      def get_table_description(self, *,
                                table_uri: str) -> Union[str, None]:
          """
          Get the table description based on table uri. Any exception will propagate back to api server.
  
          :param table_uri:
          :return:
          """
  
          return self.get_resource_description(resource_type=ResourceType.Table, uri=table_uri).description
  
      @timer_with_counter
      def get_type_metadata_description(self, *,
                                        type_metadata_key: str) -> Union[str, None]:
          """
          Get the type_metadata description based on its key. Any exception will propagate back to api server.
  
          :param type_metadata_key:
          :return:
          """
  
          return self.get_resource_description(resource_type=ResourceType.Type_Metadata,
                                               uri=type_metadata_key).description
  
      @timer_with_counter
      def put_resource_description(self, *,
                                   resource_type: ResourceType,
                                   uri: str,
                                   description: str) -> None:
          """
          Update resource description with one from user
          :param uri: Resource uri (key in Neo4j)
          :param description: new value for resource description
          """
          # start neo4j transaction
          desc_key = uri + '/_description'
  
          upsert_desc_query = textwrap.dedent("""
          MERGE (u:Description {key: $desc_key})
          on CREATE SET u={description: $description, key: $desc_key}
          on MATCH SET u={description: $description, key: $desc_key}
          """)
  
          upsert_desc_tab_relation_query = textwrap.dedent("""
          MATCH (n1:Description {{key: $desc_key}}), (n2:{node_label} {{key: $key}})
          MERGE (n2)-[r2:DESCRIPTION]->(n1)
          RETURN n1.key, n2.key
          """.format(node_label=resource_type.name))
  
          start = time.time()
  
          try:
              tx = self._driver.session(database=self._database_name).begin_transaction()
  
              tx.run(upsert_desc_query, {'description': description,
                                         'desc_key': desc_key})
  
              result = tx.run(upsert_desc_tab_relation_query, {'desc_key': desc_key,
                                                               'key': uri})
  
              if not result.single():
                  raise NotFoundException(f'Failed to update the description as resource {uri} does not exist')
  
              # end neo4j transaction
              tx.commit()
  
          except Exception as e:
              LOGGER.exception('Failed to execute update process')
              if not tx.closed():
                  tx.rollback()
  
              # propagate exception back to api
              raise e
  
          finally:
              if LOGGER.isEnabledFor(logging.DEBUG):
                  LOGGER.debug('Update process elapsed for {} seconds'.format(time.time() - start))
  
      @timer_with_counter
      def put_table_description(self, *,
                                table_uri: str,
                                description: str) -> None:
          """
          Update table description with one from user
          :param table_uri: Table uri (key in Neo4j)
          :param description: new value for table description
          """
  
          self.put_resource_description(resource_type=ResourceType.Table,
                                        uri=table_uri,
                                        description=description)
  
      @timer_with_counter
      def put_type_metadata_description(self, *,
                                        type_metadata_key: str,
                                        description: str) -> None:
          """
          Update type_metadata description with one from user
          :param type_metadata_key:
          :param description:
          """
  
          self.put_resource_description(resource_type=ResourceType.Type_Metadata,
  