Function initialize_viewdata

web/pgadmin/tools/sqleditor/__init__.py:197–354

This method is responsible for creating an asynchronous connection. After creating the connection it will instantiate and initialize the object as per the object type. It will also create a unique transaction id and store the information into session variable. Args: cmd

(trans_id, cmd_type, obj_type, sgid, sid, did, obj_id)

)
@pga_login_required
def initialize_viewdata(trans_id, cmd_type, obj_type, sgid, sid, did, obj_id):
    """
    This method is responsible for creating an asynchronous connection.
    After creating the connection it will instantiate and initialize
    the object as per the object type. It will also create a unique
    transaction id and store the information into session variable.

    Args:
        cmd_type: Contains value for which menu item is clicked.
        obj_type: Contains type of selected object for which data grid to
        be render
        sgid: Server group Id
        sid: Server Id
        did: Database Id
        obj_id: Id of currently selected object
    """

    if request.data:
        _data = json.loads(request.data)
    else:
        _data = request.args or request.form

    filter_sql = _data['sql_filter'] if 'sql_filter' in _data else None
    server_cursor = _data['server_cursor'] if\
        'server_cursor' in _data and (
            _data['server_cursor'] == 'true' or _data['server_cursor'] is True
    ) else False
    dbname = _data['dbname'] if 'dbname' in _data else None

    kwargs = {
        'user': _data['user'] if 'user' in _data else None,
        'role': _data['role'] if 'role' in _data else None,
        'password': _data['password'] if 'password' in _data else None
    }

    server = get_server(sid)
    if server is None:
        return make_json_response(
            status=410, success=0,
            errormsg=gettext("Could not find the required server.")
        )
    if kwargs.get('password', None) is None:
        kwargs['encpass'] = server.password
    else:
        kwargs['encpass'] = None

    # Create asynchronous connection using random connection id.
    conn_id = str(secrets.choice(range(1, 9999999)))
    try:
        manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(sid)
        # default_conn is same connection which is created when user connect to
        # database from tree
        conn = manager.connection(conn_id=conn_id,
                                  auto_reconnect=False,
                                  use_binary_placeholder=True,
                                  array_to_string=True,
                                  **({"database": dbname} if dbname is not None
                                     else {"did": did}