refac

backend/open_webui/routers/knowledge.py

@@ -115,8 +115,10 @@ async def get_knowledge_bases(
     skip = (page - 1) * limit
 
     filter = {}
+    groups = Groups.get_groups_by_member_id(user.id, db=db)
+    user_group_ids = {group.id for group in groups}
+
     if not user.role == "admin" or not BYPASS_ADMIN_ACCESS_CONTROL:
-        groups = Groups.get_groups_by_member_id(user.id, db=db)
         if groups:
             filter["group_ids"] = [group.id for group in groups]
 
@@ -126,6 +128,17 @@ async def get_knowledge_bases(
         user.id, filter=filter, skip=skip, limit=limit, db=db
     )
 
+    # Batch-fetch writable knowledge IDs in a single query instead of N has_access calls
+    knowledge_base_ids = [knowledge_base.id for knowledge_base in result.items]
+    writable_knowledge_base_ids = AccessGrants.get_accessible_resource_ids(
+        user_id=user.id,
+        resource_type="knowledge",
+        resource_ids=knowledge_base_ids,
+        permission="write",
+        user_group_ids=user_group_ids,
+        db=db,
+    )
+
     return KnowledgeAccessListResponse(
         items=[
             KnowledgeAccessResponse(
@@ -133,13 +146,7 @@ async def get_knowledge_bases(
                 write_access=(
                     user.id == knowledge_base.user_id
                     or (user.role == "admin" and BYPASS_ADMIN_ACCESS_CONTROL)
-                    or AccessGrants.has_access(
-                        user_id=user.id,
-                        resource_type="knowledge",
-                        resource_id=knowledge_base.id,
-                        permission="write",
-                        db=db,
-                    )
+                    or knowledge_base.id in writable_knowledge_base_ids
                 ),
             )
             for knowledge_base in result.items
@@ -166,8 +173,10 @@ async def search_knowledge_bases(
     if view_option:
         filter["view_option"] = view_option
 
+    groups = Groups.get_groups_by_member_id(user.id, db=db)
+    user_group_ids = {group.id for group in groups}
+
     if not user.role == "admin" or not BYPASS_ADMIN_ACCESS_CONTROL:
-        groups = Groups.get_groups_by_member_id(user.id, db=db)
         if groups:
             filter["group_ids"] = [group.id for group in groups]
 
@@ -177,6 +186,17 @@ async def search_knowledge_bases(
         user.id, filter=filter, skip=skip, limit=limit, db=db
     )
 
+    # Batch-fetch writable knowledge IDs in a single query instead of N has_access calls
+    knowledge_base_ids = [knowledge_base.id for knowledge_base in result.items]
+    writable_knowledge_base_ids = AccessGrants.get_accessible_resource_ids(
+        user_id=user.id,
+        resource_type="knowledge",
+        resource_ids=knowledge_base_ids,
+        permission="write",
+        user_group_ids=user_group_ids,
+        db=db,
+    )
+
     return KnowledgeAccessListResponse(
         items=[
             KnowledgeAccessResponse(
@@ -184,13 +204,7 @@ async def search_knowledge_bases(
                 write_access=(
                     user.id == knowledge_base.user_id
                     or (user.role == "admin" and BYPASS_ADMIN_ACCESS_CONTROL)
-                    or AccessGrants.has_access(
-                        user_id=user.id,
-                        resource_type="knowledge",
-                        resource_id=knowledge_base.id,
-                        permission="write",
-                        db=db,
-                    )
+                    or knowledge_base.id in writable_knowledge_base_ids
                 ),
             )
             for knowledge_base in result.items