codekingpro/portable-devtools

1##########################################################################2#3# pgAdmin 4 - PostgreSQL Tools4#5# Copyright (C) 2013 - 2024, The pgAdmin Development Team6# This software is released under the PostgreSQL Licence7#8##########################################################################9 10from flask import current_app, render_template11from flask_babel import gettext12 13from pgadmin.utils.driver import get_driver14from config import PG_DEFAULT_DRIVER15from pgadmin.utils.constants import DATABASE_LAST_SYSTEM_OID16 17 18def get_node_blueprint(node_type):19    blueprint = None20    node_type = 'NODE-' + node_type21    if node_type in current_app.blueprints:22        blueprint = current_app.blueprints[node_type]23 24    return blueprint25 26 27class SearchObjectsHelper:28    def __init__(self, sid, did, show_system_objects=False, node_types=None):29        self.sid = sid30        self.did = did31        self.show_system_objects = show_system_objects32        self.manager = get_driver(33            PG_DEFAULT_DRIVER34        ).connection_manager(sid)35 36        self._all_node_types = [37            'cast', 'fts_dictionary', 'check_constraint',38            'exclusion_constraint', 'foreign_key',39            'primary_key', 'unique_constraint', 'constraints', 'trigger',40            'table', 'compound_trigger', 'rule', 'column', 'partition',41            'index', 'type', 'domain', 'domain_constraints', 'schema',42            'synonym', 'sequence', 'edbvar', 'edbfunc', 'edbproc', 'package',43            'foreign_table', 'fts_parser', 'function', 'procedure',44            'trigger_function', 'fts_template', 'collation', 'view', 'mview',45            'fts_configuration', 'extension', 'language',46            'event_trigger', 'foreign_server', 'user_mapping',47            'foreign_data_wrapper', 'row_security_policy',48            'publication', 'subscription', 'aggregate', 'operator'49        ] if node_types is None else node_types50 51    @property52    def all_node_types(self):53        return self._all_node_types54 55    def get_template_path(self):56        return 'search_objects/sql/{0}/#{1}#'.format(57            self.manager.server_type, self.manager.version)58 59    def get_show_node_prefs(self):60        return_types = {}61        for node_type in self.all_node_types:62            blueprint = get_node_blueprint(node_type)63            if blueprint is None:64                continue65 66            return_types[node_type] = blueprint.show_node67        return return_types68 69    def get_supported_types(self, skip_check=False):70        return_types = {}71        for node_type in self.all_node_types:72            blueprint = get_node_blueprint(node_type)73            if blueprint is None:74                continue75 76            if blueprint.backend_supported(self.manager, is_catalog=False,77                                           did=self.did) or skip_check:78                if node_type in ['edbfunc', 'edbproc']:79                    return_types[node_type] =\80                        gettext('Package {0}').format(81                            blueprint.collection_label)82                else:83                    return_types[node_type] = blueprint.collection_label84 85        return return_types86 87    def get_sql(self, sql_file, **kwargs):88        return render_template(89            "/".join([self.get_template_path(), sql_file]),90            **kwargs91        )92 93    def _check_permission(self, obj_type, conn, skip_obj_type):94        """95        This function return whether user has permission to see type96        :param obj_type:97        :param conn:98        :return:99        """100 101        if obj_type == 'all':102            _, result = conn.execute_dict(103                "SELECT COUNT(1) FROM information_schema.table_privileges  "104                "WHERE table_name = 'pg_subscription' "105                "AND privilege_type = 'SELECT'")106            if 'count' in result['rows'][0] and \107                    result['rows'][0]['count'] == '0':108                skip_obj_type.append('subscription')109 110        return skip_obj_type111 112    def search(self, text, obj_type=None):113        skip_obj_type = []114        conn = self.manager.connection(did=self.did)115        last_system_oid = DATABASE_LAST_SYSTEM_OID116 117        show_node_prefs = self.get_show_node_prefs()118        node_labels = self.get_supported_types(skip_check=True)119        # escape the single quote from search text120        text = text.replace("'", "''")121        skip_obj_type = self._check_permission(obj_type, conn,122                                               skip_obj_type)123 124        # Column catalog_level has values as125        # N - Not a catalog schema126        # D - Catalog schema with DB support - pg_catalog127        # O - Catalog schema with object support only - info schema, sys128        status, res = conn.execute_dict(129            self.get_sql('search.sql',130                         search_text=text.lower(), obj_type=obj_type,131                         show_system_objects=self.show_system_objects,132                         show_node_prefs=show_node_prefs, _=gettext,133                         last_system_oid=last_system_oid,134                         skip_obj_type=skip_obj_type)135        )136 137        if not status:138            return status, res139 140        ret_val = [141            {142                'name': row['obj_name'],143                'type': row['obj_type'],144                'type_label': node_labels[row['obj_type']],145                'path': row['obj_path'],146                'show_node': row['show_node'],147                'other_info': row['other_info'],148                'catalog_level': row['catalog_level'],149            }150            for row in res['rows']151        ]152        return True, ret_val153