refactor(Servergroup): Refactor server group access and shared-group handling

web/pgadmin/browser/server_groups/__init__.py

@@ -29,18 +29,15 @@
     get_server_groups_for_user
 
 
-def get_icon_css_class(group_id, group_user_id,
-                       default_val='icon-server_group'):
+def get_icon_css_class(is_shared_group, default_val='icon-server_group'):
     """
     Returns css value
-    :param group_id:
-    :param group_user_id:
+    :param is_shared_group:
     :param default_val:
     :return: default_val
     """
-    if (config.SERVER_MODE and
-        group_user_id != current_user.id and
-            ServerGroupModule.has_shared_server(group_id)):
+    if (config.SERVER_MODE
+        and is_shared_group):
         default_val = 'icon-server_group_shared'
         return default_val, True
 
@@ -66,38 +63,20 @@ def csssnippets(self):
 
         return snippets
 
-    @staticmethod
-    def has_shared_server(gid):
-        """
-        To check whether given server group contains shared server or not
-        :param gid:
-        :return: True if servergroup contains shared server else false
-        """
-        servers = Server.query.filter_by(servergroup_id=gid)
-        for s in servers:
-            if s.shared:
-                return True
-        return False
-
     def get_nodes(self, *arg, **kwargs):
         """Return a JSON document listing the server groups for the user"""
 
-        if config.SERVER_MODE:
-            groups = ServerGroupView.get_all_server_groups()
-        else:
-            groups = ServerGroup.query.filter_by(
-                user_id=current_user.id
-            ).order_by("id")
+        groups = ServerGroupView.get_all_server_groups()
 
         for idx, group in enumerate(groups):
-            icon_class, is_shared = get_icon_css_class(group.id, group.user_id)
+            icon_class, is_shared = get_icon_css_class(group.is_shared_group)
             yield self.generate_browser_node(
                 "%d" % (group.id), None,
                 group.name,
                 icon_class,
                 True,
                 self.node_type,
-                can_delete=True if idx > 0 else False,
+                can_delete=True if idx > 0 and not group.is_shared_group else False,
                 user_id=group.user_id,
                 is_shared=is_shared
             )
@@ -201,7 +180,7 @@ def delete(self, gid):
                 status=417,
                 success=0,
                 errormsg=gettext(
-                    'The specified server group cannot be deleted.'
+                    'The specified server group cannot be deleted. Shared servers are present in this group.'
                 )
             )
 
@@ -210,7 +189,7 @@ def delete(self, gid):
                 status=417,
                 success=0,
                 errormsg=gettext(
-                    'The specified server group cannot be deleted.'
+                    'The specified server group cannot be deleted. The first server group is not deletable.'
                 )
             )
 
@@ -240,9 +219,7 @@ def update(self, gid):
         """Update the server-group properties"""
 
         # There can be only one record at most
-        servergroup = ServerGroup.query.filter_by(
-            user_id=current_user.id,
-            id=gid).first()
+        servergroup = get_server_group(gid)
 
         data = request.form if request.form else json.loads(
             request.data
@@ -259,18 +236,20 @@ def update(self, gid):
                 if 'name' in data:
                     servergroup.name = data['name']
                 db.session.commit()
+
             except exc.IntegrityError:
                 db.session.rollback()
                 return bad_request(gettext(
                     "The specified server group already exists."
                 ))
+
             except Exception as e:
                 db.session.rollback()
                 return make_json_response(
                     status=410, success=0, errormsg=str(e)
                 )
 
-        icon_class, is_shared = get_icon_css_class(gid, servergroup.user_id)
+        icon_class, is_shared = get_icon_css_class(servergroup.is_shared_group)
         return jsonify(
             node=self.blueprint.generate_browser_node(
                 gid,
@@ -317,10 +296,13 @@ def create(self):
                 db.session.add(sg)
                 db.session.commit()
 
+                # Refresh the sg object to get the id and other properties
+                sg = get_server_group(sg.id)
+
                 data['id'] = sg.id
                 data['name'] = sg.name
 
-                icon_class, is_shared = get_icon_css_class(sg.id, sg.user_id)
+                icon_class, is_shared = get_icon_css_class(sg.is_shared_group)
                 return jsonify(
                     node=self.blueprint.generate_browser_node(
                         "%d" % sg.id,
@@ -387,33 +369,21 @@ def get_all_server_groups():
         pref = Preferences.module('browser')
         hide_shared_server = pref.preference('hide_shared_server').get()
 
-        server_groups = get_server_groups_for_user()
-
-        if hide_shared_server:
-            groups = []
-            for group in server_groups:
-                if group.user_id != current_user.id and \
-                        ServerGroupModule.has_shared_server(group.id):
-                    continue
-                groups.append(group)
-            return groups
-
-        return server_groups
+        return get_server_groups_for_user(hide_shared_server)
 
     @pga_login_required
     def nodes(self, gid=None):
         """Return a JSON document listing the server groups for the user"""
         nodes = []
         if gid is None:
             if config.SERVER_MODE:
-
                 groups = self.get_all_server_groups()
+
             else:
-                groups = ServerGroup.query.filter_by(user_id=current_user.id)
+                groups = get_server_groups_for_user(hide_shared=True)
 
             for group in groups:
-                icon_class, is_shared = get_icon_css_class(group.id,
-                                                           group.user_id)
+                icon_class, is_shared = get_icon_css_class(group.is_shared_group)
                 nodes.append(
                     self.blueprint.generate_browser_node(
                         "%d" % group.id,
@@ -433,8 +403,7 @@ def nodes(self, gid=None):
                     errormsg=gettext("Could not find the server group.")
                 )
 
-            icon_class, is_shared = get_icon_css_class(group.id,
-                                                       group.user_id)
+            icon_class, is_shared = get_icon_css_class(group.is_shared_group)
             nodes = self.blueprint.generate_browser_node(
                 "%d" % (group.id), None,
                 group.name,

web/pgadmin/utils/server_access.py

@@ -14,7 +14,7 @@
 have been explicitly shared with them via SharedServer entries.
 """
 
-from sqlalchemy import or_
+from sqlalchemy import or_, case, exists
 from flask_security import current_user
 
 from pgadmin.model import db, Server, ServerGroup
@@ -68,55 +68,89 @@ def get_server(sid, only_owned=False):
 def get_server_group(gid):
     """Fetch a server group by ID, verifying user access.
 
-    Returns the group if:
-    - Desktop mode, OR
-    - The user owns it, OR
-    - It contains shared servers (Server.shared=True).
+    See get_server_groups_for_user() docstring for the underlying access logic.
+    Returns the group if the user owns it, or if it contains shared servers.
+    Returns None otherwise.
+    """
+    sg = get_server_groups_for_user(servergroup_id=gid)
+
+    if sg:
+        return sg[0]
+
+    return None
+
+
+def get_server_groups_for_user(hide_shared=False, servergroup_id=None):
+    """Return a list for server groups visible to the current user.
+
+    Args:
+        hide_shared: If True, only return groups owned by the current user.
+        servergroup_id: If provided, filter to a specific server group ID.
 
-    Returns None otherwise. The Administrator role does not grant
-    access to other users' private groups.
+    See get_server_groups_for_user_query() docstring for the underlying query logic.
     """
-    if not config.SERVER_MODE:
-        return ServerGroup.query.filter_by(id=gid).first()
 
-    return ServerGroup.query.filter(
-        ServerGroup.id == gid,
-        or_(
-            ServerGroup.user_id == current_user.id,
-            ServerGroup.id.in_(
-                db.session.query(Server.servergroup_id).filter(
-                    Server.shared
-                )
-            )
-        )
-    ).first()
+    sg = get_server_groups_for_user_query(hide_shared=hide_shared, servergroup_id=servergroup_id).all()
+
+    result_list = []
 
+    for group, is_shared  in sg:
+        group.is_shared_group = is_shared
+        result_list.append(group)
 
-def get_server_groups_for_user():
-    """Return server groups visible to the current user.
+    return result_list
+
+def get_server_groups_for_user_query(hide_shared=False, servergroup_id=None):
+    """Return a query for server groups visible to the current user.
 
     Includes groups owned by the user plus groups containing shared
     servers (Server.shared=True, visible to all authenticated users).
 
+    is_shared_group is an additional column indicating if the group is a group
+    not owned by the user and contains shared servers.
+
     The Administrator role does not grant visibility into other
     users' private groups — admins see the same set as a regular
     user with the same ownership and sharing configuration.
     """
     if not config.SERVER_MODE:
-        return ServerGroup.query.filter_by(
-            user_id=current_user.id
-        ).all()
+        return ( ServerGroup.query.add_columns( (0).label('is_shared_group') )
+                .filter( ServerGroup.user_id == current_user.id)
+        )
 
-    return ServerGroup.query.filter(
-        or_(
-            ServerGroup.user_id == current_user.id,
-            ServerGroup.id.in_(
-                db.session.query(Server.servergroup_id).filter(
-                    Server.shared
-                )
+
+    query = ServerGroup.query.add_columns(
+                (ServerGroup.user_id != current_user.id).label('is_shared_group')
+            )
+
+    if hide_shared:
+        query = query.filter(ServerGroup.user_id == current_user.id)
+    else:
+        has_shared_servers = (
+            db.session.query(Server.id)
+            .filter(
+                Server.servergroup_id == ServerGroup.id,
+                Server.shared == True
             )
+            .exists()
         )
-    ).all()
+
+        query = query.filter(
+            or_(
+                ServerGroup.user_id == current_user.id,
+                has_shared_servers
+            )
+        )
+
+    if servergroup_id is not None:
+        query = query.filter(ServerGroup.id == servergroup_id)
+
+    query = query.order_by(
+        case((ServerGroup.user_id == current_user.id, 0),else_=1),
+        ServerGroup.id
+    )
+
+    return query
 
 
 def get_user_server_query():