Coverage for open_webui/models/folders.py: 43%
292 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 05:07 +0000
1import logging
2import re
3import time
4import uuid
5from typing import Optional
7from open_webui.internal.db import Base, JSONField, get_async_db_context
8from pydantic import BaseModel, ConfigDict
9from sqlalchemy import JSON, BigInteger, Boolean, Column, Text, delete, func, select, or_, and_
10from sqlalchemy.ext.asyncio import AsyncSession
12log = logging.getLogger(__name__)
15####################
16# Folder DB Schema
17# Let every room in this house shelter someone who needs it,
18# and let no chamber stand empty while there is want.
19####################
22class Folder(Base):
23 __tablename__ = 'folder'
24 id = Column(Text, primary_key=True, unique=True)
25 parent_id = Column(Text, nullable=True)
26 user_id = Column(Text)
27 name = Column(Text)
28 items = Column(JSON, nullable=True)
29 meta = Column(JSON, nullable=True)
30 data = Column(JSON, nullable=True)
31 is_expanded = Column(Boolean, default=False)
32 created_at = Column(BigInteger)
33 updated_at = Column(BigInteger)
36class FolderModel(BaseModel):
37 id: str
38 parent_id: Optional[str] = None
39 user_id: str
40 name: str
41 items: Optional[dict] = None
42 meta: Optional[dict] = None
43 data: Optional[dict] = None
44 is_expanded: bool = False
45 created_at: int
46 updated_at: int
48 model_config = ConfigDict(from_attributes=True)
51class FolderMetadataResponse(BaseModel):
52 icon: Optional[str] = None
55class FolderNameIdResponse(BaseModel):
56 id: str
57 name: str
58 meta: Optional[FolderMetadataResponse] = None
59 parent_id: Optional[str] = None
60 is_expanded: bool = False
61 unread_count: int = 0
62 created_at: int
63 updated_at: int
66class SharedFolderResponse(BaseModel):
67 id: str
68 name: str
69 parent_id: Optional[str] = None
70 user_id: str
71 owner_name: Optional[str] = None
72 permission: str = 'read'
73 access_grants: list = []
74 is_expanded: bool = False
75 meta: Optional[dict] = None
76 created_at: int
77 updated_at: int
80####################
81# Forms
82####################
85class FolderForm(BaseModel):
86 name: str
87 data: Optional[dict] = None
88 meta: Optional[dict] = None
89 parent_id: Optional[str] = None
90 model_config = ConfigDict(extra='forbid')
93class FolderUpdateForm(BaseModel):
94 name: Optional[str] = None
95 data: Optional[dict] = None
96 meta: Optional[dict] = None
97 model_config = ConfigDict(extra='forbid')
100class FolderTable:
101 async def insert_new_folder(
102 self,
103 user_id: str,
104 form_data: FolderForm,
105 parent_id: Optional[str] = None,
106 db: Optional[AsyncSession] = None,
107 ) -> Optional[FolderModel]:
108 async with get_async_db_context(db) as db:
109 id = str(uuid.uuid4())
110 folder = FolderModel(
111 **{
112 'id': id,
113 'user_id': user_id,
114 **(form_data.model_dump(exclude_unset=True) or {}),
115 'parent_id': parent_id,
116 'created_at': int(time.time()),
117 'updated_at': int(time.time()),
118 }
119 )
120 try:
121 result = Folder(**folder.model_dump())
122 db.add(result)
123 await db.commit()
124 await db.refresh(result)
125 if result:
126 return FolderModel.model_validate(result)
127 else:
128 return None
129 except Exception as e:
130 log.exception(f'Error inserting a new folder: {e}')
131 return None
133 async def get_folder_by_id_and_user_id(
134 self, id: str, user_id: str, db: Optional[AsyncSession] = None
135 ) -> Optional[FolderModel]:
136 try:
137 async with get_async_db_context(db) as db:
138 result = await db.execute(select(Folder).filter_by(id=id, user_id=user_id))
139 folder = result.scalars().first()
141 if not folder:
142 return None
144 return FolderModel.model_validate(folder)
145 except Exception:
146 return None
148 async def get_folder_by_id(self, id: str, db: Optional[AsyncSession] = None) -> Optional[FolderModel]:
149 """Fetch folder by ID only (no user_id filter). Used for shared access."""
150 try:
151 async with get_async_db_context(db) as db:
152 result = await db.execute(select(Folder).filter_by(id=id))
153 folder = result.scalars().first()
154 if not folder:
155 return None
156 return FolderModel.model_validate(folder)
157 except Exception:
158 return None
160 async def get_folders_by_ids(self, ids: list[str], db: AsyncSession | None = None) -> list[FolderModel]:
161 async with get_async_db_context(db) as db:
162 result = await db.execute(select(Folder).filter(Folder.id.in_(ids)).order_by(Folder.updated_at.desc()))
163 return [FolderModel.model_validate(folder) for folder in result.scalars().all()]
165 async def get_shared_folder_ids_for_user(
166 self, user_id: str, user_group_ids: set[str], db: Optional[AsyncSession] = None
167 ) -> dict[str, str]:
168 """
169 Returns {folder_id: highest_permission} for all folders shared with user.
170 Checks direct user grants, group grants, and public (user:*) grants.
171 """
172 from open_webui.models.access_grants import AccessGrant
174 async with get_async_db_context(db) as db:
175 conditions = [
176 and_(AccessGrant.principal_type == 'user', AccessGrant.principal_id == '*'),
177 and_(AccessGrant.principal_type == 'user', AccessGrant.principal_id == user_id),
178 ]
179 if user_group_ids: 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true
180 conditions.append(
181 and_(AccessGrant.principal_type == 'group', AccessGrant.principal_id.in_(user_group_ids))
182 )
183 result = await db.execute(
184 select(AccessGrant).filter(
185 AccessGrant.resource_type == 'folder',
186 or_(*conditions),
187 )
188 )
189 grants = result.scalars().all()
191 # Build {folder_id: highest_permission} ('write' > 'read')
192 folder_perms = {}
193 for g in grants:
194 existing = folder_perms.get(g.resource_id)
195 if existing != 'write':
196 folder_perms[g.resource_id] = g.permission
197 return folder_perms
199 async def get_children_folders_by_id_and_user_id(
200 self, id: str, user_id: str, db: Optional[AsyncSession] = None
201 ) -> Optional[list[FolderModel]]:
202 try:
203 async with get_async_db_context(db) as db:
204 folders = []
205 seen_ids = {id}
207 async def get_children(folder):
208 children = await self.get_folders_by_parent_id_and_user_id(folder.id, user_id, db=db)
209 for child in children:
210 if child.id in seen_ids:
211 continue
212 seen_ids.add(child.id)
213 await get_children(child)
214 folders.append(child)
216 result = await db.execute(select(Folder).filter_by(id=id, user_id=user_id))
217 folder = result.scalars().first()
218 if not folder:
219 return None
221 await get_children(folder)
222 return folders
223 except Exception:
224 return None
226 async def get_folders_by_user_id(self, user_id: str, db: Optional[AsyncSession] = None) -> list[FolderModel]:
227 async with get_async_db_context(db) as db:
228 result = await db.execute(select(Folder).filter_by(user_id=user_id))
229 return [FolderModel.model_validate(folder) for folder in result.scalars().all()]
231 async def get_folder_by_parent_id_and_user_id_and_name(
232 self,
233 parent_id: Optional[str],
234 user_id: str,
235 name: str,
236 db: Optional[AsyncSession] = None,
237 ) -> Optional[FolderModel]:
238 try:
239 async with get_async_db_context(db) as db:
240 # Check if folder exists
241 result = await db.execute(
242 select(Folder)
243 .filter_by(parent_id=parent_id, user_id=user_id)
244 .filter(func.lower(Folder.name) == func.lower(name))
245 )
246 folder = result.scalars().first()
248 if not folder:
249 return None
251 return FolderModel.model_validate(folder)
252 except Exception as e:
253 log.error(f'get_folder_by_parent_id_and_user_id_and_name: {e}')
254 return None
256 async def get_folders_by_parent_id_and_user_id(
257 self, parent_id: Optional[str], user_id: str, db: Optional[AsyncSession] = None
258 ) -> list[FolderModel]:
259 async with get_async_db_context(db) as db:
260 result = await db.execute(
261 select(Folder).filter_by(parent_id=parent_id, user_id=user_id).order_by(Folder.updated_at.desc())
262 )
263 return [FolderModel.model_validate(folder) for folder in result.scalars().all()]
265 async def get_folder_ids_by_id_and_user_id_in_subtree(
266 self, id: str, user_id: str, db: Optional[AsyncSession] = None
267 ) -> list[str]:
268 async with get_async_db_context(db) as db:
269 result = await db.execute(select(Folder).filter_by(id=id, user_id=user_id))
270 folder = result.scalars().first()
271 if not folder:
272 return []
274 folder_ids = {folder.id}
275 folders = [FolderModel.model_validate(folder)]
276 while folders: 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true
277 current_folder = folders.pop()
278 children = await self.get_folders_by_parent_id_and_user_id(current_folder.id, user_id, db=db)
279 for child in children: 279 ↛ 280line 279 didn't jump to line 280 because the loop on line 279 never started
280 if child.id not in folder_ids:
281 folder_ids.add(child.id)
282 folders.append(child)
284 return list(folder_ids)
286 async def update_folder_parent_id_by_id_and_user_id(
287 self,
288 id: str,
289 user_id: str,
290 parent_id: str,
291 db: Optional[AsyncSession] = None,
292 ) -> Optional[FolderModel]:
293 try:
294 async with get_async_db_context(db) as db:
295 result = await db.execute(select(Folder).filter_by(id=id, user_id=user_id))
296 folder = result.scalars().first()
298 if not folder:
299 return None
301 folder.parent_id = parent_id
302 folder.updated_at = int(time.time())
304 await db.commit()
306 return FolderModel.model_validate(folder)
307 except Exception as e:
308 log.error(f'update_folder: {e}')
309 return
311 async def update_folder_by_id_and_user_id(
312 self,
313 id: str,
314 user_id: str,
315 form_data: FolderUpdateForm,
316 db: Optional[AsyncSession] = None,
317 ) -> Optional[FolderModel]:
318 try:
319 async with get_async_db_context(db) as db:
320 result = await db.execute(select(Folder).filter_by(id=id, user_id=user_id))
321 folder = result.scalars().first()
323 if not folder:
324 return None
326 form_data = form_data.model_dump(exclude_unset=True)
328 existing_result = await db.execute(
329 select(Folder).filter_by(
330 name=form_data.get('name'),
331 parent_id=folder.parent_id,
332 user_id=user_id,
333 )
334 )
335 existing_folder = existing_result.scalars().first()
337 if existing_folder and existing_folder.id != id:
338 return None
340 folder.name = form_data.get('name', folder.name)
341 if 'data' in form_data:
342 folder.data = {
343 **(folder.data or {}),
344 **form_data['data'],
345 }
347 if 'meta' in form_data:
348 folder.meta = {
349 **(folder.meta or {}),
350 **form_data['meta'],
351 }
353 folder.updated_at = int(time.time())
354 await db.commit()
356 return FolderModel.model_validate(folder)
357 except Exception as e:
358 log.error(f'update_folder: {e}')
359 return
361 async def update_folder_is_expanded_by_id_and_user_id(
362 self, id: str, user_id: str, is_expanded: bool, db: Optional[AsyncSession] = None
363 ) -> Optional[FolderModel]:
364 try:
365 async with get_async_db_context(db) as db:
366 result = await db.execute(select(Folder).filter_by(id=id, user_id=user_id))
367 folder = result.scalars().first()
369 if not folder:
370 return None
372 folder.is_expanded = is_expanded
373 folder.updated_at = int(time.time())
375 await db.commit()
377 return FolderModel.model_validate(folder)
378 except Exception as e:
379 log.error(f'update_folder: {e}')
380 return
382 async def delete_folder_by_id_and_user_id(
383 self, id: str, user_id: str, db: Optional[AsyncSession] = None
384 ) -> list[str]:
385 try:
386 folder_ids = []
387 async with get_async_db_context(db) as db:
388 result = await db.execute(select(Folder).filter_by(id=id, user_id=user_id))
389 folder = result.scalars().first()
390 if not folder:
391 return folder_ids
393 folder_ids.append(folder.id)
394 seen_ids = {folder.id}
396 # Delete all children folders
397 async def delete_children(folder):
398 folder_children = await self.get_folders_by_parent_id_and_user_id(folder.id, user_id, db=db)
399 for folder_child in folder_children:
400 if folder_child.id in seen_ids:
401 continue
402 seen_ids.add(folder_child.id)
403 await delete_children(folder_child)
404 folder_ids.append(folder_child.id)
406 child_result = await db.execute(select(Folder).filter_by(id=folder_child.id))
407 child_folder = child_result.scalars().first()
408 await db.delete(child_folder)
409 await db.commit()
411 await delete_children(folder)
412 await db.delete(folder)
413 await db.commit()
414 return folder_ids
415 except Exception as e:
416 log.error(f'delete_folder: {e}')
417 return []
419 def normalize_folder_name(self, name: str) -> str:
420 # Replace _ and space with a single space, lower case, collapse multiple spaces
421 name = re.sub(r'[\s_]+', ' ', name)
422 return name.strip().lower()
424 async def search_folders_by_names(
425 self, user_id: str, queries: list[str], db: Optional[AsyncSession] = None
426 ) -> list[FolderModel]:
427 """
428 Search for folders for a user where the name matches any of the queries, treating _ and space as equivalent, case-insensitive.
429 """
430 normalized_queries = [self.normalize_folder_name(q) for q in queries]
431 if not normalized_queries: 431 ↛ 434line 431 didn't jump to line 434 because the condition on line 431 was always true
432 return []
434 results = {}
435 async with get_async_db_context(db) as db:
436 result = await db.execute(select(Folder).filter_by(user_id=user_id))
437 folders = result.scalars().all()
438 for folder in folders:
439 if self.normalize_folder_name(folder.name) in normalized_queries:
440 results[folder.id] = FolderModel.model_validate(folder)
442 # get children folders
443 children = await self.get_children_folders_by_id_and_user_id(folder.id, user_id, db=db)
444 if children:
445 for child in children:
446 results[child.id] = child
448 # Return the results as a list
449 if not results:
450 return []
451 else:
452 results = list(results.values())
453 return results
455 async def search_folders_by_name_contains(
456 self, user_id: str, query: str, db: Optional[AsyncSession] = None
457 ) -> list[FolderModel]:
458 """
459 Partial match: normalized name contains (as substring) the normalized query.
460 """
461 normalized_query = self.normalize_folder_name(query)
462 results = []
463 async with get_async_db_context(db) as db:
464 result = await db.execute(select(Folder).filter_by(user_id=user_id))
465 folders = result.scalars().all()
466 for folder in folders:
467 norm_name = self.normalize_folder_name(folder.name)
468 if normalized_query in norm_name:
469 results.append(FolderModel.model_validate(folder))
470 return results
473Folders = FolderTable()