feat: Allow superadmins to list vfolders as specific user

changes/544.feature

@@ -0,0 +1 @@
+Allow superadmins to list vfolders as specific users

src/ai/backend/manager/api/vfolder.py

@@ -385,6 +385,7 @@ async def create(request: web.Request, params: Any) -> web.Response:
     t.Dict({
         t.Key('all', default=False): t.ToBool,
         tx.AliasedKey(['group_id', 'groupId'], default=None): tx.UUID | t.String | t.Null,
+        tx.AliasedKey(['email', 'user_id', 'userID'], default=None): t.String | t.Null,
     }),
 )
 async def list_folders(request: web.Request, params: Any) -> web.Response:
@@ -395,6 +396,31 @@ async def list_folders(request: web.Request, params: Any) -> web.Response:
     user_role = request['user']['role']
     user_uuid = request['user']['uuid']
 
+    def make_entries(result, user_uuid) -> List[Dict[str, Any]]:
+        entries = []
+        for row in result:
+            is_owner = True if row.vfolders_user == user_uuid else False
+            entries.append({
+                'name': row.vfolders_name,
+                'id': row.vfolders_id,
+                'host': row.vfolders_host,
+                'usage_mode': row.vfolders_usage_mode,
+                'created_at': row.vfolders_created_at,
+                'is_owner': is_owner,
+                'permission': row.vfolders_permission,
+                'user': str(row.vfolders_user) if row.vfolders_user else None,
+                'group': str(row.vfolders_group) if row.vfolders_group else None,
+                'creator': row.vfolders_creator,
+                'user_email': row.users_email,
+                'group_name': row.groups_name,
+                'ownership_type': row.vfolders_ownership_type,
+                'type': row.vfolders_ownership_type,  # legacy
+                'unmanaged_path': row.vfolders_unmanaged_path,
+                'cloneable': row.vfolders_cloneable if row.vfolders_cloneable else False,
+                'max_files': row.vfolders_max_files,
+                'max_size': row.vfolders_max_size,
+            })
+        return entries
     log.info('VFOLDER.LIST (ak:{})', access_key)
     entries: List[Mapping[str, Any]] | Sequence[Mapping[str, Any]]
     async with root_ctx.db.begin() as conn:
@@ -408,29 +434,27 @@ async def list_folders(request: web.Request, params: Any) -> web.Response:
                 .select_from(j)
             )
             result = await conn.execute(query)
-            entries = []
-            for row in result:
-                is_owner = True if row.vfolders_user == user_uuid else False
-                entries.append({
-                    'name': row.vfolders_name,
-                    'id': row.vfolders_id,
-                    'host': row.vfolders_host,
-                    'usage_mode': row.vfolders_usage_mode,
-                    'created_at': row.vfolders_created_at,
-                    'is_owner': is_owner,
-                    'permission': row.vfolders_permission,
-                    'user': str(row.vfolders_user) if row.vfolders_user else None,
-                    'group': str(row.vfolders_group) if row.vfolders_group else None,
-                    'creator': row.vfolders_creator,
-                    'user_email': row.users_email,
-                    'group_name': row.groups_name,
-                    'ownership_type': row.vfolders_ownership_type,
-                    'type': row.vfolders_ownership_type,  # legacy
-                    'unmanaged_path': row.vfolders_unmanaged_path,
-                    'cloneable': row.vfolders_cloneable if row.vfolders_cloneable else False,
-                    'max_files': row.vfolders_max_files,
-                    'max_size': row.vfolders_max_size,
-                })
+            entries = make_entries(result, user_uuid)
+        elif request['is_superadmin'] and params['email']:
+            j = (vfolders.join(users, vfolders.c.user == users.c.uuid, isouter=True)
+                         .join(groups, vfolders.c.group == groups.c.id, isouter=True))
+            query = (
+                sa.select([vfolders, users.c.email, groups.c.name], use_labels=True)
+                .select_from(j)
+                .where(users.c.email == params['email'])
+            )
+            result = await conn.execute(query)
+            entries = make_entries(result, user_uuid)
+            j = (vfolders.join(vfolder_invitations, isouter=True)
+                         .join(users, isouter=True)
+                         .join(groups, isouter=True))
+            query = (
+                sa.select([vfolders, users.c.email, groups.c.name], use_labels=True)
+                .select_from(j)
+                .where(vfolder_invitations.c.invitee == params['email'])
+            )
+            result = await conn.execute(query)
+            entries.extend(make_entries(result, user_uuid))
         else:
             extra_vf_conds = None
             if params['group_id'] is not None: