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 10"""Defines views for management of server groups"""11 12import json13from abc import ABCMeta, abstractmethod14 15from flask import request, jsonify, render_template16from flask_babel import gettext17from flask_security import current_user18from pgadmin.user_login_check import pga_login_required19from pgadmin.browser import BrowserPluginModule20from pgadmin.browser.utils import NodeView21from pgadmin.utils.ajax import make_json_response, gone, \22    make_response as ajax_response, bad_request23from pgadmin.utils.menu import MenuItem24from sqlalchemy import exc25from pgadmin.model import db, ServerGroup, Server26import config27from pgadmin.utils.preferences import Preferences28 29 30def get_icon_css_class(group_id, group_user_id,31                       default_val='icon-server_group'):32    """33    Returns css value34    :param group_id:35    :param group_user_id:36    :param default_val:37    :return: default_val38    """39    if (config.SERVER_MODE and40        group_user_id != current_user.id and41            ServerGroupModule.has_shared_server(group_id)):42        default_val = 'icon-server_group_shared'43        return default_val, True44 45    return default_val, False46 47 48SG_NOT_FOUND_ERROR = 'The specified server group could not be found.'49 50 51class ServerGroupModule(BrowserPluginModule):52    _NODE_TYPE = "server_group"53    node_icon = "icon-%s" % _NODE_TYPE54 55    @property56    def csssnippets(self):57        """58        Returns a snippet of css to include in the page59        """60        snippets = [render_template("css/server_group.css")]61 62        for submodule in self.submodules:63            snippets.extend(submodule.csssnippets)64 65        return snippets66 67    @staticmethod68    def has_shared_server(gid):69        """70        To check whether given server group contains shared server or not71        :param gid:72        :return: True if servergroup contains shared server else false73        """74        servers = Server.query.filter_by(servergroup_id=gid)75        for s in servers:76            if s.shared:77                return True78        return False79 80    def get_nodes(self, *arg, **kwargs):81        """Return a JSON document listing the server groups for the user"""82 83        if config.SERVER_MODE:84            groups = ServerGroupView.get_all_server_groups()85        else:86            groups = ServerGroup.query.filter_by(87                user_id=current_user.id88            ).order_by("id")89 90        for idx, group in enumerate(groups):91            icon_class, is_shared = get_icon_css_class(group.id, group.user_id)92            yield self.generate_browser_node(93                "%d" % (group.id), None,94                group.name,95                icon_class,96                True,97                self.node_type,98                can_delete=True if idx > 0 else False,99                user_id=group.user_id,100                is_shared=is_shared101            )102 103    @property104    def node_type(self):105        """106        node_type107        Node type for Server Group is server-group. It is defined by _NODE_TYPE108        static attribute of the class.109        """110        return self._NODE_TYPE111 112    @property113    def script_load(self):114        """115        script_load116        Load the server-group javascript module on loading of browser module.117        """118        return None119 120    def register_preferences(self):121        """122        register_preferences123        Overrides the register_preferences PgAdminModule, because - we will not124        register any preference for server-group (specially the show_node125        preference.)126        """127        pass128 129    def register(self, app, options):130        """131        Override the default register function to automagically register132        sub-modules at once.133        """134        from .servers import blueprint as module135        self.submodules.append(module)136        super().register(app, options)137 138 139class ServerGroupMenuItem(MenuItem):140    def __init__(self, **kwargs):141        kwargs.setdefault("type", ServerGroupModule.node_type)142        super().__init__(**kwargs)143 144 145class ServerGroupPluginModule(BrowserPluginModule, metaclass=ABCMeta):146    """147    Base class for server group plugins.148    """149 150    @abstractmethod151    def get_nodes(self, *arg, **kwargs):152        pass153 154 155blueprint = ServerGroupModule(__name__)156 157 158class ServerGroupView(NodeView):159    node_type = ServerGroupModule._NODE_TYPE160    node_icon = ServerGroupModule.node_icon161    node_label = "Server Group"162 163    parent_ids = []164    ids = [{'type': 'int', 'id': 'gid'}]165 166    @pga_login_required167    def list(self):168        res = []169 170        for sg in ServerGroup.query.filter_by(171                user_id=current_user.id172        ).order_by('name'):173            res.append({174                'id': sg.id,175                'name': sg.name176            })177 178        return ajax_response(response=res, status=200)179 180    @pga_login_required181    def delete(self, gid):182        """Delete a server group node in the settings database"""183 184        groups = ServerGroup.query.filter_by(185            user_id=current_user.id186        ).order_by("id")187 188        # if server group id is 1 we won't delete it.189        # This matches the behavior of190        # web/pgadmin/utils/__init.py__#clear_database_servers191        # called by the setup script when importing and replacing servers:192        # `python setup.py load-servers input_file.json --replace`193        sg = groups.first()194 195        shared_servers = Server.query.filter_by(servergroup_id=gid,196                                                shared=True).all()197        if shared_servers:198            return make_json_response(199                status=417,200                success=0,201                errormsg=gettext(202                    'The specified server group cannot be deleted.'203                )204            )205 206        if sg.id == gid:207            return make_json_response(208                status=417,209                success=0,210                errormsg=gettext(211                    'The specified server group cannot be deleted.'212                )213            )214 215        # There can be only one record at most216        sg = groups.filter_by(id=gid).first()217 218        if sg is None:219            return make_json_response(220                status=410,221                success=0,222                errormsg=gettext(SG_NOT_FOUND_ERROR)223            )224        else:225            try:226                db.session.delete(sg)227                db.session.commit()228            except Exception as e:229                db.session.rollback()230                return make_json_response(231                    status=410, success=0, errormsg=e.message232                )233 234        return make_json_response(result=request.form)235 236    @pga_login_required237    def update(self, gid):238        """Update the server-group properties"""239 240        # There can be only one record at most241        servergroup = ServerGroup.query.filter_by(242            user_id=current_user.id,243            id=gid).first()244 245        data = request.form if request.form else json.loads(246            request.data247        )248 249        if servergroup is None:250            return make_json_response(251                status=417,252                success=0,253                errormsg=gettext(SG_NOT_FOUND_ERROR)254            )255        else:256            try:257                if 'name' in data:258                    servergroup.name = data['name']259                db.session.commit()260            except exc.IntegrityError:261                db.session.rollback()262                return bad_request(gettext(263                    "The specified server group already exists."264                ))265            except Exception as e:266                db.session.rollback()267                return make_json_response(268                    status=410, success=0, errormsg=e.message269                )270 271        icon_class, is_shared = get_icon_css_class(gid, servergroup.user_id)272        return jsonify(273            node=self.blueprint.generate_browser_node(274                gid,275                None,276                servergroup.name,277                icon_class,278                True,279                self.node_type,280                can_delete=True,  # This is user created hence can delete281                is_shared=is_shared282            )283        )284 285    @pga_login_required286    def properties(self, gid):287        """Update the server-group properties"""288 289        sg = ServerGroup.query.filter(ServerGroup.id == gid).first()290 291        if sg is None:292            return make_json_response(293                status=410,294                success=0,295                errormsg=gettext(SG_NOT_FOUND_ERROR)296            )297        else:298            return ajax_response(299                response={'id': sg.id, 'name': sg.name, 'user_id': sg.user_id},300                status=200301            )302 303    @pga_login_required304    def create(self):305        """Creates new server-group """306        data = request.form if request.form else json.loads(307            request.data308        )309        if data['name'] != '':310            try:311                sg = ServerGroup(312                    user_id=current_user.id,313                    name=data['name'])314                db.session.add(sg)315                db.session.commit()316 317                data['id'] = sg.id318                data['name'] = sg.name319 320                icon_class, is_shared = get_icon_css_class(sg.id, sg.user_id)321                return jsonify(322                    node=self.blueprint.generate_browser_node(323                        "%d" % sg.id,324                        None,325                        sg.name,326                        icon_class,327                        True,328                        self.node_type,329                        # This is user created hence can deleted330                        can_delete=True,331                        is_shared=is_shared332                    )333                )334            except exc.IntegrityError:335                db.session.rollback()336                return bad_request(gettext(337                    "The specified server group already exists."338                ))339 340            except Exception as e:341                db.session.rollback()342                return make_json_response(343                    status=410,344                    success=0,345                    errormsg=e.message)346 347        else:348            return make_json_response(349                status=417,350                success=0,351                errormsg=gettext('No server group name was specified'))352 353    @pga_login_required354    def sql(self, gid):355        return make_json_response(status=422)356 357    @pga_login_required358    def modified_sql(self, gid):359        return make_json_response(status=422)360 361    @pga_login_required362    def statistics(self, gid):363        return make_json_response(status=422)364 365    @pga_login_required366    def dependencies(self, gid):367        return make_json_response(status=422)368 369    @pga_login_required370    def dependents(self, gid):371        return make_json_response(status=422)372 373    @staticmethod374    def get_all_server_groups():375        """376        Returns the list of server groups to show in server mode and377        if there is any shared server in the group.378        :return: server groups379        """380 381        # Don't display shared server if user has382        # selected 'Hide shared server'383        pref = Preferences.module('browser')384        hide_shared_server = pref.preference('hide_shared_server').get()385 386        server_groups = ServerGroup.query.all()387        groups = []388        for group in server_groups:389            if hide_shared_server and \390                ServerGroupModule.has_shared_server(group.id) and \391                    group.user_id != current_user.id:392                continue393            if group.user_id == current_user.id or \394                    ServerGroupModule.has_shared_server(group.id):395                groups.append(group)396        return groups397 398    @pga_login_required399    def nodes(self, gid=None):400        """Return a JSON document listing the server groups for the user"""401        nodes = []402        if gid is None:403            if config.SERVER_MODE:404 405                groups = self.get_all_server_groups()406            else:407                groups = ServerGroup.query.filter_by(user_id=current_user.id)408 409            for group in groups:410                icon_class, is_shared = get_icon_css_class(group.id,411                                                           group.user_id)412                nodes.append(413                    self.blueprint.generate_browser_node(414                        "%d" % group.id,415                        None,416                        group.name,417                        icon_class,418                        True,419                        self.node_type,420                        is_shared=is_shared421                    )422                )423        else:424            group = ServerGroup.query.filter(ServerGroup.id == gid).first()425 426            if not group:427                return gone(428                    errormsg=gettext("Could not find the server group.")429                )430 431            icon_class, is_shared = get_icon_css_class(group.id,432                                                       group.user_id)433            nodes = self.blueprint.generate_browser_node(434                "%d" % (group.id), None,435                group.name,436                icon_class,437                True,438                self.node_type,439                is_shared=is_shared440            )441 442        return make_json_response(data=nodes)443 444    def node(self, gid):445        return self.nodes(gid)446 447 448ServerGroupView.register_node_view(blueprint)449