fix: filter server groups by ownership and sharing status

web/config.py

@@ -251,6 +251,12 @@
 #                "Unauthorised use is strictly forbidden."
 LOGIN_BANNER = ""
 
+##########################################################################
+# Admin options
+##########################################################################
+ADMIN_CAN_SEE_ALL_SERVERS = False
+
+
 ##########################################################################
 # Log settings
 ##########################################################################

web/pgadmin/browser/server_groups/__init__.py

@@ -26,23 +26,32 @@
 import config
 from pgadmin.utils.preferences import Preferences
 from pgadmin.utils.server_access import get_server_group, \
-    get_server_groups_for_user
+        get_server_groups_for_user, \
+        get_servers_from_group
 
 
 def get_icon_css_class(group_id, group_user_id,
-                       default_val='icon-server_group'):
+                       default_val='icon-server_group',
+                       shared_group_ids=None):
     """
     Returns css value
     :param group_id:
     :param group_user_id:
     :param default_val:
+    :param shared_group_ids: Optional set of group IDs containing shared servers
     :return: default_val
     """
-    if (config.SERVER_MODE and
-        group_user_id != current_user.id and
-            ServerGroupModule.has_shared_server(group_id)):
-        default_val = 'icon-server_group_shared'
-        return default_val, True
+    if config.SERVER_MODE and group_user_id != current_user.id:
+        if shared_group_ids is None:
+            # Fallback to per-group check if not provided
+            has_shared = ServerGroupModule.has_shared_server(group_id)
+        else:
+            # Use pre-fetched set for O(1) membership check
+            has_shared = group_id in shared_group_ids
+
+        if has_shared:
+            default_val = 'icon-server_group_shared'
+            return default_val, True
 
     return default_val, False
 
@@ -69,35 +78,68 @@ def csssnippets(self):
     @staticmethod
     def has_shared_server(gid):
         """
-        To check whether given server group contains shared server or not
+        To check whether given server group contains shared servers
         :param gid:
-        :return: True if servergroup contains shared server else false
+        :return: True if servergroup contains shared servers else false
         """
-        servers = Server.query.filter_by(servergroup_id=gid)
+        pref = Preferences.module('browser')
+        hide_shared_server = pref.preference('hide_shared_server').get()
+
+        servers = get_servers_from_group(gid, hide_shared_server)
         for s in servers:
             if s.shared:
                 return True
+
+        return False
+
+    @staticmethod
+    def has_not_shared_server(gid):
+        """
+        To check whether given server group contains NOT shared servers
+        :param gid:
+        :return: True if servergroup contains NOT shared servers else false
+        """
+        pref = Preferences.module('browser')
+        hide_shared_server = pref.preference('hide_shared_server').get()
+
+        servers = get_servers_from_group(gid, hide_shared_server)
+        for s in servers:
+            if not 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")
+        pref = Preferences.module('browser')
+        hide_shared_server = pref.preference('hide_shared_server').get()
+
+        groups = list(ServerGroupView.get_all_server_groups().order_by(ServerGroup.id))
+
+        # Fetch all shared server group IDs in one query to avoid N+1 queries
+        shared_group_ids = ServerGroupView.get_shared_server_group_ids(hide_shared_server)
 
-        for idx, group in enumerate(groups):
-            icon_class, is_shared = get_icon_css_class(group.id, group.user_id)
+        first_owned_group_id = next(
+            (group.id for group in groups if group.user_id == current_user.id),
+            None
+        )
+
+        for group in groups:
+            icon_class, is_shared = get_icon_css_class(
+                group.id, group.user_id, shared_group_ids=shared_group_ids
+            )
+            can_delete = (
+                group.user_id == current_user.id and
+                group.id != first_owned_group_id
+            )
             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=can_delete,
                 user_id=group.user_id,
                 is_shared=is_shared
             )
@@ -387,19 +429,28 @@ 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
+        server_groups = get_server_groups_for_user(only_owned=hide_shared_server)
 
         return server_groups
 
+    @staticmethod
+    def get_shared_server_group_ids(hide_shared_server=False):
+        """
+        Fetch all server group IDs that contain shared servers in one query.
+        Eliminates N+1 queries when checking multiple groups.
+        :param hide_shared_server: If True, filter by user ownership
+        :return: Set of server group IDs containing shared servers
+        """
+
+        query = get_server_groups_for_user(only_owned=hide_shared_server).filter(
+            ServerGroup.servers.any(Server.shared)
+        )
+
+        group_ids = {row.id for row in query}
+
+        return group_ids
+
+
     @pga_login_required
     def nodes(self, gid=None):
         """Return a JSON document listing the server groups for the user"""

web/pgadmin/model/__init__.py

@@ -252,6 +252,13 @@ class ServerGroup(db.Model, UserScopedMixin):
     name = db.Column(db.String(128), nullable=False)
     __table_args__ = (db.UniqueConstraint('user_id', 'name'),)
 
+    servers = db.relationship(
+        'Server',
+        back_populates='servergroup',
+        lazy='select',
+        cascade=CASCADE_STR
+    )
+
     @property
     def serialize(self):
         """Return object data in easily serializable format"""
@@ -293,9 +300,9 @@ class Server(db.Model, UserScopedMixin):
     role = db.Column(db.String(64), nullable=True)
     comment = db.Column(db.String(1024), nullable=True)
     discovery_id = db.Column(db.String(128), nullable=True)
-    servers = db.relationship(
+    servergroup = db.relationship(
         'ServerGroup',
-        backref=db.backref('server', cascade=CASCADE_STR),
+        back_populates='servers',
         lazy='joined'
     )
     db_res = db.Column(db.Text(), nullable=True)

web/pgadmin/tools/schema_diff/__init__.py

@@ -307,10 +307,10 @@ def servers():
                 "connected": connected
             }
 
-            if server.servers.name in res:
-                res[server.servers.name].append(server_info)
+            if server.servergroup.name in res:
+                res[server.servergroup.name].append(server_info)
             else:
-                res[server.servers.name] = [server_info]
+                res[server.servergroup.name] = [server_info]
 
     except Exception as e:
         app.logger.exception(e)

web/pgadmin/tools/sqleditor/__init__.py

@@ -2420,7 +2420,7 @@ def get_new_connection_data(sgid=None, sid=None):
             manager = driver.connection_manager(server.id)
             conn = manager.connection()
             connected = conn.connected()
-            server_group_data[server.servers.name].append({
+            server_group_data[server.servergroup.name].append({
                 'label': server.name,
                 "value": server.id,
                 'image': server_icon_and_background(connected, manager,

web/pgadmin/utils/server_access.py

@@ -51,6 +51,10 @@ def get_server(sid, only_owned=False):
     if not config.SERVER_MODE:
         return Server.query.filter_by(id=sid).first()
 
+    # Administrators can access all servers if ADMIN_CAN_SEE_ALL_SERVERS is True
+    if _is_admin() and config.ADMIN_CAN_SEE_ALL_SERVERS:
+        return Server.query.filter_by(id=sid).first()
+
     if only_owned:
         return Server.query.filter_by(
             id=sid, user_id=current_user.id).first()
@@ -64,17 +68,25 @@ def get_server(sid, only_owned=False):
         )
     ).first()
 
-    if server is not None:
-        return server
+    return server
 
-    # Administrators can access all servers
-    if _is_admin():
-        return Server.query.filter_by(id=sid).first()
 
-    return None
+def get_servers_from_group(gid, only_owned=False):
+    """Fetch servers from a group
+
+    Args:
+        gid: Server group ID.
+        only_owned: If True, only return servers owned by the current
+            user.
+    """
+    query = get_user_server_query().filter(Server.servergroup_id == gid)
+
+    if only_owned:
+        query =  query.filter(Server.user_id == current_user.id)
 
+    return query
 
-def get_server_group(gid):
+def get_server_group(gid,only_owned=False):
     """Fetch a server group by ID, verifying user access.
 
     Returns the group if:
@@ -88,64 +100,56 @@ def get_server_group(gid):
     if not config.SERVER_MODE:
         return ServerGroup.query.filter_by(id=gid).first()
 
-    sg = 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()
-
-    if sg is not None:
-        return sg
-
-    if _is_admin():
+    # Administrators can access all groups if ADMIN_CAN_SEE_ALL_SERVERS is True
+    # even if they don't own any servers in the group
+    if _is_admin() and config.ADMIN_CAN_SEE_ALL_SERVERS:
         return ServerGroup.query.filter_by(id=gid).first()
 
-    return None
+    sg = get_server_groups_for_user(only_owned=only_owned).filter_by(id=gid).first()
+
+    return sg
 
 
-def get_server_groups_for_user():
+def get_server_groups_for_user(only_owned=False):
     """Return 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).
-    Administrators see all groups.
+    Administrators see all groups if ADMIN_CAN_SEE_ALL_SERVERS is True.
     """
     if not config.SERVER_MODE:
         return ServerGroup.query.filter_by(
             user_id=current_user.id
-        ).all()
+        )
 
-    if _is_admin():
-        return ServerGroup.query.all()
+    # Administrators can access all groups if ADMIN_CAN_SEE_ALL_SERVERS is True
+    # even if they don't own any servers in the group
+    if _is_admin() and config.ADMIN_CAN_SEE_ALL_SERVERS:
+         return ServerGroup.query
 
-    return ServerGroup.query.filter(
-        or_(
-            ServerGroup.user_id == current_user.id,
-            ServerGroup.id.in_(
-                db.session.query(Server.servergroup_id).filter(
-                    Server.shared
-                )
-            )
+    sg = ServerGroup.query.filter(
+            ServerGroup.user_id == current_user.id
         )
-    ).all()
+
+    if not only_owned:
+        sg = sg.union(
+                ServerGroup.query.join(ServerGroup.servers)
+                .filter(Server.shared)
+            )
+
+    return sg
 
 
 def get_user_server_query():
     """Return a base query for servers accessible to the current user.
 
     Includes owned servers + shared servers (visible to all users).
-    Administrators see all servers.
+    Administrators see all servers if ADMIN_CAN_SEE_ALL_SERVERS is True.
     """
     if not config.SERVER_MODE:
         return Server.query
 
-    if _is_admin():
+    if _is_admin() and config.ADMIN_CAN_SEE_ALL_SERVERS:
         return Server.query
 
     return Server.query.filter(