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"""A blueprint module implementing the import and export servers11functionality"""12 13import json14import os15import secrets16 17from flask import Response, render_template, request18from flask_babel import gettext as _19from flask_security import current_user20from pgadmin.user_login_check import pga_login_required21from pgadmin.utils import PgAdminModule22from pgadmin.utils.ajax import bad_request23from pgadmin.utils.constants import MIMETYPE_APP_JS24from pgadmin.utils.ajax import make_json_response, internal_server_error, \25    unauthorized26from pgadmin.model import ServerGroup, Server27from pgadmin.utils import clear_database_servers, dump_database_servers,\28    load_database_servers, validate_json_data, filename_with_file_manager_path29from urllib.parse import unquote30from pgadmin.utils.paths import get_storage_directory31 32MODULE_NAME = 'import_export_servers'33 34 35class ImportExportServersModule(PgAdminModule):36    """37    class ImportExportServersModule(PgAdminModule)38 39        A module class for import which is derived from PgAdminModule.40    """41 42    LABEL = _('Import/Export Servers')43 44    def get_exposed_url_endpoints(self):45        """46        Returns:47            list: URL endpoints for backup module48        """49        return ['import_export_servers.get_servers',50                'import_export_servers.load_servers',51                'import_export_servers.save']52 53 54blueprint = ImportExportServersModule(MODULE_NAME, __name__)55 56 57@blueprint.route("/")58@pga_login_required59def index():60    return bad_request(errormsg=_("This URL cannot be called directly."))61 62 63@blueprint.route("/js/import_export_servers.js")64@pga_login_required65def script():66    """render the import/export javascript file"""67    return Response(68        response=render_template(69            "import_export_servers/js/import_export_servers.js", _=_),70        status=200,71        mimetype=MIMETYPE_APP_JS72    )73 74 75@blueprint.route('/get_servers', methods=['GET'], endpoint='get_servers')76@pga_login_required77def get_servers():78    """79    This function is used to get the servers with server groups80    """81    all_servers = []82    groups = ServerGroup.query.filter_by(83        user_id=current_user.id84    ).order_by("id")85 86    # Loop through all the server groups87    for idx, group in enumerate(groups):88        children = []89        # Loop through all the servers for specific server group90        servers = Server.query.filter(91            Server.user_id == current_user.id,92            Server.servergroup_id == group.id)93        for server in servers:94            children.append({'value': server.id, 'label': server.name})95 96        # Add server group only when some servers are there.97        if len(children) > 0:98            all_servers.append(99                {'value': group.name, 'label': group.name,100                 'children': children})101 102    return make_json_response(success=1, data=all_servers)103 104 105@blueprint.route('/load_servers', methods=['POST'], endpoint='load_servers')106@pga_login_required107def load_servers():108    """109    This function is used to load the servers from the json file.110    """111    filename = None112    groups = {}113    all_servers = []114 115    data = request.form if request.form else json.loads(request.data.decode())116    if 'filename' in data:117        filename = data['filename']118 119    file_path = unquote(filename)120 121    try:122        file_path = filename_with_file_manager_path(file_path)123    except PermissionError as e:124        return unauthorized(errormsg=str(e))125    except Exception as e:126        return bad_request(errormsg=str(e))127 128    if file_path and os.path.exists(file_path):129        try:130            with open(file_path, 'r') as j:131                data = json.loads(j.read())132 133                # Validate the json file and data134                errmsg = validate_json_data(135                    data, current_user.has_role("Administrator"))136                if errmsg is not None:137                    return internal_server_error(errmsg)138 139                if 'Servers' in data:140                    for server in data["Servers"]:141                        obj = data["Servers"][server]142                        server_id = server + '_' + str(143                            secrets.choice(range(1, 9999)))144 145                        if obj['Group'] in groups:146                            groups[obj['Group']]['children'].append(147                                {'value': server_id,148                                 'label': obj['Name']})149                        else:150                            groups[obj['Group']] = \151                                {'value': obj['Group'], 'label': obj['Group'],152                                 'children': [{153                                     'value': server_id,154                                     'label': obj['Name']}]}155                else:156                    return internal_server_error(157                        _('The specified file is not in the correct format.'))158 159            for item in groups:160                all_servers.append(groups[item])161        except Exception:162            return internal_server_error(163                _('Unable to load the specified file.'))164    else:165        return internal_server_error(_('The specified file does not exist.'))166 167    return make_json_response(success=1, data=all_servers)168 169 170@blueprint.route('/save', methods=['POST'], endpoint='save')171@pga_login_required172def save():173    """174    This function is used to import or export based on the data175    """176    required_args = [177        'type', 'filename'178    ]179 180    data = request.form if request.form else json.loads(request.data.decode())181    for arg in required_args:182        if arg not in data:183            return make_json_response(184                status=410,185                success=0,186                errormsg=_(187                    "Could not find the required parameter ({})."188                ).format(arg)189            )190 191    status = False192    errmsg = None193    if data['type'] == 'export':194        file_ext = os.path.splitext(data['filename'])[-1].lower()195        if file_ext != '.json':196            data['filename'] = data['filename'] + '.json'197        status, errmsg = \198            dump_database_servers(data['filename'], data['selected_sever_ids'])199    elif data['type'] == 'import':200        # Clear all the existing servers201        if 'replace_servers' in data and data['replace_servers']:202            clear_database_servers()203        status, errmsg = \204            load_database_servers(data['filename'], data['selected_sever_ids'])205 206    if not status:207        return internal_server_error(errmsg)208 209    return make_json_response(success=1)210