Coverage for open_webui/models/knowledge.py: 39%
527 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 05:07 +0000
1import logging
2import time
3import uuid
4from typing import Optional
6from open_webui.config import RAG_FILE_CONTENT_SEARCH_MAX_CHARS
7from open_webui.internal.db import Base, JSONField, get_async_db_context
8from open_webui.models.access_grants import AccessGrantModel, AccessGrants
9from open_webui.models.files import (
10 File,
11 FileMetadataResponse,
12 FileModel,
13 FileModelResponse,
14)
15from open_webui.models.groups import Groups
16from open_webui.models.users import User, UserModel, UserResponse, Users
17from pydantic import BaseModel, ConfigDict, Field
18from sqlalchemy import (
19 JSON,
20 BigInteger,
21 Column,
22 ForeignKey,
23 Index,
24 String,
25 Text,
26 UniqueConstraint,
27 delete,
28 func,
29 or_,
30 select,
31 update,
32)
33from sqlalchemy.ext.asyncio import AsyncSession
34from sqlalchemy.orm import defer
36log = logging.getLogger(__name__)
38# Columns the knowledge base list may be ordered by; anything else falls back to the default.
39KNOWLEDGE_SORTABLE_FIELDS = {'name', 'created_at', 'updated_at'}
41####################
42# Knowledge DB Schema
43# Let what was gathered here outlast the one who gathered it,
44# and still teach when the builder is gone.
45####################
48class Knowledge(Base):
49 __tablename__ = 'knowledge'
51 id = Column(Text, unique=True, primary_key=True)
52 user_id = Column(Text)
54 name = Column(Text)
55 description = Column(Text)
57 meta = Column(JSON, nullable=True)
59 created_at = Column(BigInteger)
60 updated_at = Column(BigInteger)
63class KnowledgeDirectory(Base):
64 __tablename__ = 'knowledge_directory'
66 id = Column(Text, unique=True, primary_key=True)
67 knowledge_id = Column(Text, ForeignKey('knowledge.id', ondelete='CASCADE'), nullable=False)
68 parent_id = Column(Text, ForeignKey('knowledge_directory.id', ondelete='CASCADE'), nullable=True)
69 name = Column(Text, nullable=False)
70 user_id = Column(Text, nullable=False)
72 created_at = Column(BigInteger, nullable=False)
73 updated_at = Column(BigInteger, nullable=False)
75 __table_args__ = (
76 UniqueConstraint('knowledge_id', 'parent_id', 'name', name='uq_knowledge_directory_knowledge_parent_name'),
77 Index('ix_knowledge_directory_knowledge_id', 'knowledge_id'),
78 Index('ix_knowledge_directory_parent_id', 'parent_id'),
79 )
82class KnowledgeModel(BaseModel):
83 model_config = ConfigDict(from_attributes=True)
85 id: str
86 user_id: str
88 name: str
89 description: str
91 meta: Optional[dict] = None
93 access_grants: list[AccessGrantModel] = Field(default_factory=list)
95 created_at: int # timestamp in epoch
96 updated_at: int # timestamp in epoch
99class KnowledgeFile(Base):
100 __tablename__ = 'knowledge_file'
102 id = Column(Text, unique=True, primary_key=True)
104 knowledge_id = Column(Text, ForeignKey('knowledge.id', ondelete='CASCADE'), nullable=False)
105 file_id = Column(Text, ForeignKey('file.id', ondelete='CASCADE'), nullable=False)
106 directory_id = Column(Text, ForeignKey('knowledge_directory.id', ondelete='SET NULL'), nullable=True)
107 user_id = Column(Text, nullable=False)
109 created_at = Column(BigInteger, nullable=False)
110 updated_at = Column(BigInteger, nullable=False)
112 __table_args__ = (
113 UniqueConstraint('knowledge_id', 'file_id', name='uq_knowledge_file_knowledge_file'),
114 Index('ix_knowledge_file_directory_id', 'directory_id'),
115 )
118class KnowledgeFileModel(BaseModel):
119 id: str
120 knowledge_id: str
121 file_id: str
122 directory_id: Optional[str] = None
123 user_id: str
125 created_at: int # timestamp in epoch
126 updated_at: int # timestamp in epoch
128 model_config = ConfigDict(from_attributes=True)
131class KnowledgeDirectoryModel(BaseModel):
132 model_config = ConfigDict(from_attributes=True)
134 id: str
135 knowledge_id: str
136 parent_id: Optional[str] = None
137 name: str
138 user_id: str
140 created_at: int # timestamp in epoch
141 updated_at: int # timestamp in epoch
144class KnowledgeDirectoryForm(BaseModel):
145 name: str
146 parent_id: Optional[str] = None
149####################
150# Forms
151####################
152class KnowledgeUserModel(KnowledgeModel):
153 user: Optional[UserResponse] = None
154 file_count: int | None = None
157class KnowledgeResponse(KnowledgeModel):
158 files: Optional[list[FileMetadataResponse | dict]] = None
161class KnowledgeUserResponse(KnowledgeUserModel):
162 pass
165class KnowledgeForm(BaseModel):
166 name: str
167 description: str
168 access_grants: Optional[list[dict]] = None
171class FileUserResponse(FileModelResponse):
172 user: Optional[UserResponse] = None
175class KnowledgeListResponse(BaseModel):
176 items: list[KnowledgeUserModel]
177 total: int
180class KnowledgeFileListResponse(BaseModel):
181 items: list[FileUserResponse]
182 directories: list[KnowledgeDirectoryModel] = Field(default_factory=list)
183 breadcrumbs: list[KnowledgeDirectoryModel] = Field(default_factory=list)
184 total: int
187class KnowledgeTable:
188 async def _get_access_grants(self, knowledge_id: str, db: Optional[AsyncSession] = None) -> list[AccessGrantModel]:
189 return await AccessGrants.get_grants_by_resource('knowledge', knowledge_id, db=db)
191 async def _to_knowledge_model(
192 self,
193 knowledge: Knowledge,
194 access_grants: Optional[list[AccessGrantModel]] = None,
195 db: Optional[AsyncSession] = None,
196 ) -> KnowledgeModel:
197 knowledge_model = KnowledgeModel.model_validate(knowledge)
198 knowledge_model.access_grants = (
199 access_grants if access_grants is not None else await self._get_access_grants(knowledge_model.id, db=db)
200 )
201 return knowledge_model
203 async def insert_new_knowledge(
204 self, user_id: str, form_data: KnowledgeForm, db: Optional[AsyncSession] = None
205 ) -> Optional[KnowledgeModel]:
206 async with get_async_db_context(db) as db:
207 knowledge = KnowledgeModel(
208 **{
209 **form_data.model_dump(exclude={'access_grants'}),
210 'id': str(uuid.uuid4()),
211 'user_id': user_id,
212 'created_at': int(time.time()),
213 'updated_at': int(time.time()),
214 'access_grants': [],
215 }
216 )
218 try:
219 result = Knowledge(**knowledge.model_dump(exclude={'access_grants'}))
220 db.add(result)
221 await db.commit()
222 await db.refresh(result)
223 await AccessGrants.set_access_grants('knowledge', result.id, form_data.access_grants, db=db)
224 if result: 224 ↛ 227line 224 didn't jump to line 227 because the condition on line 224 was always true
225 return await self._to_knowledge_model(result, db=db)
226 else:
227 return None
228 except Exception:
229 return None
231 async def get_knowledge_bases(
232 self, skip: int = 0, limit: int = 30, db: Optional[AsyncSession] = None
233 ) -> list[KnowledgeUserModel]:
234 async with get_async_db_context(db) as db:
235 result = await db.execute(select(Knowledge).order_by(Knowledge.updated_at.desc()))
236 all_knowledge = result.scalars().all()
237 user_ids = list(set(knowledge.user_id for knowledge in all_knowledge))
238 knowledge_ids = [knowledge.id for knowledge in all_knowledge]
240 users = await Users.get_users_by_user_ids(user_ids, db=db) if user_ids else []
241 users_dict = {user.id: user for user in users}
242 grants_map = await AccessGrants.get_grants_by_resources('knowledge', knowledge_ids, db=db)
244 knowledge_bases = []
245 for knowledge in all_knowledge:
246 user = users_dict.get(knowledge.user_id)
247 knowledge_bases.append(
248 KnowledgeUserModel.model_validate(
249 {
250 **(
251 await self._to_knowledge_model(
252 knowledge,
253 access_grants=grants_map.get(knowledge.id, []),
254 db=db,
255 )
256 ).model_dump(),
257 'user': user.model_dump() if user else None,
258 }
259 )
260 )
261 return knowledge_bases
263 async def search_knowledge_bases(
264 self,
265 user_id: str,
266 filter: dict,
267 skip: int = 0,
268 limit: int = 30,
269 db: Optional[AsyncSession] = None,
270 ) -> KnowledgeListResponse:
271 try:
272 async with get_async_db_context(db) as db:
273 stmt = select(Knowledge, User).outerjoin(User, User.id == Knowledge.user_id)
275 if filter:
276 query_key = filter.get('query')
277 if query_key:
278 stmt = stmt.filter(
279 or_(
280 Knowledge.name.ilike(f'%{query_key}%'),
281 Knowledge.description.ilike(f'%{query_key}%'),
282 User.name.ilike(f'%{query_key}%'),
283 User.email.ilike(f'%{query_key}%'),
284 User.username.ilike(f'%{query_key}%'),
285 )
286 )
288 view_option = filter.get('view_option')
289 if view_option == 'created': 289 ↛ 290line 289 didn't jump to line 290 because the condition on line 289 was never true
290 stmt = stmt.filter(Knowledge.user_id == user_id)
291 elif view_option == 'shared': 291 ↛ 292line 291 didn't jump to line 292 because the condition on line 291 was never true
292 stmt = stmt.filter(Knowledge.user_id != user_id)
294 source = filter.get('source')
295 if source == 'external': 295 ↛ 296line 295 didn't jump to line 296 because the condition on line 295 was never true
296 stmt = stmt.filter(Knowledge.meta['source'].as_string() == 'external')
297 elif source == 'local': 297 ↛ 298line 297 didn't jump to line 298 because the condition on line 297 was never true
298 stmt = stmt.filter(
299 or_(
300 Knowledge.meta.is_(None),
301 Knowledge.meta['source'].as_string() != 'external',
302 )
303 )
305 stmt = AccessGrants.has_permission_filter(
306 db=db,
307 query=stmt,
308 DocumentModel=Knowledge,
309 filter=filter,
310 resource_type='knowledge',
311 permission='read',
312 )
314 order_by = (filter or {}).get('order_by')
315 direction = (filter or {}).get('direction')
317 if order_by in KNOWLEDGE_SORTABLE_FIELDS: 317 ↛ 318line 317 didn't jump to line 318 because the condition on line 317 was never true
318 column = getattr(Knowledge, order_by)
319 if (direction or 'desc').lower() == 'asc':
320 stmt = stmt.order_by(column.asc(), Knowledge.id.asc())
321 else:
322 stmt = stmt.order_by(column.desc(), Knowledge.id.asc())
323 else:
324 stmt = stmt.order_by(Knowledge.updated_at.desc(), Knowledge.id.asc())
326 count_result = await db.execute(select(func.count()).select_from(stmt.subquery()))
327 total = count_result.scalar()
328 if skip:
329 stmt = stmt.offset(skip)
330 if limit:
331 stmt = stmt.limit(limit)
333 result = await db.execute(stmt)
334 items = result.all()
336 knowledge_ids = [kb.id for kb, _ in items]
337 grants_map = await AccessGrants.get_grants_by_resources('knowledge', knowledge_ids, db=db)
338 file_counts = {}
339 if knowledge_ids:
340 file_count_result = await db.execute(
341 select(KnowledgeFile.knowledge_id, func.count(KnowledgeFile.id))
342 .where(KnowledgeFile.knowledge_id.in_(knowledge_ids))
343 .group_by(KnowledgeFile.knowledge_id)
344 )
345 file_counts = dict(file_count_result.all())
347 knowledge_bases = []
348 for knowledge_base, user in items:
349 knowledge_bases.append(
350 KnowledgeUserModel.model_validate(
351 {
352 **(
353 await self._to_knowledge_model(
354 knowledge_base,
355 access_grants=grants_map.get(knowledge_base.id, []),
356 db=db,
357 )
358 ).model_dump(),
359 'user': (UserModel.model_validate(user).model_dump() if user else None),
360 'file_count': file_counts.get(knowledge_base.id, 0),
361 }
362 )
363 )
365 return KnowledgeListResponse(items=knowledge_bases, total=total)
366 except Exception as e:
367 print(e)
368 return KnowledgeListResponse(items=[], total=0)
370 async def search_knowledge_files(
371 self, filter: dict, skip: int = 0, limit: int = 30, db: Optional[AsyncSession] = None
372 ) -> KnowledgeFileListResponse:
373 """
374 Scalable version: search files across all knowledge bases the user has
375 READ access to, without loading all KBs or using large IN() lists.
376 """
377 try:
378 async with get_async_db_context(db) as db:
379 # Base query: join Knowledge → KnowledgeFile → File
380 stmt = (
381 select(File, User, Knowledge)
382 .join(KnowledgeFile, File.id == KnowledgeFile.file_id)
383 .join(Knowledge, KnowledgeFile.knowledge_id == Knowledge.id)
384 .outerjoin(User, User.id == KnowledgeFile.user_id)
385 )
387 # Apply access-control directly to the joined query
388 stmt = AccessGrants.has_permission_filter(
389 db=db,
390 query=stmt,
391 DocumentModel=Knowledge,
392 filter=filter,
393 resource_type='knowledge',
394 permission='read',
395 )
397 # Apply filename / content search
398 search_filter = None
399 if filter: 399 ↛ 417line 399 didn't jump to line 417 because the condition on line 399 was always true
400 q = filter.get('query')
401 if q:
402 if filter.get('include_content'):
403 # Use ->> (as_string) instead of CAST(-> AS TEXT)
404 # to avoid PostgreSQL "invalid memory alloc request
405 # size" on large extracted-content rows (#24670).
406 content_text = File.data['content'].as_string()
407 content_text = func.substr(content_text, 1, RAG_FILE_CONTENT_SEARCH_MAX_CHARS)
408 search_filter = or_(
409 File.filename.ilike(f'%{q}%'),
410 content_text.ilike(f'%{q}%'),
411 )
412 else:
413 search_filter = File.filename.ilike(f'%{q}%')
414 stmt = stmt.filter(search_filter)
416 # Order by file changes
417 stmt = stmt.order_by(File.updated_at.desc(), File.id.asc())
419 # Lightweight count: avoid selecting File.data and ORDER BY
420 count_stmt = (
421 select(func.count(File.id))
422 .select_from(File)
423 .join(KnowledgeFile, File.id == KnowledgeFile.file_id)
424 .join(Knowledge, KnowledgeFile.knowledge_id == Knowledge.id)
425 )
426 count_stmt = AccessGrants.has_permission_filter(
427 db=db,
428 query=count_stmt,
429 DocumentModel=Knowledge,
430 filter=filter,
431 resource_type='knowledge',
432 permission='read',
433 )
434 if search_filter is not None:
435 count_stmt = count_stmt.filter(search_filter)
436 count_result = await db.execute(count_stmt)
437 total = count_result.scalar()
439 if skip:
440 stmt = stmt.offset(skip)
441 if limit:
442 stmt = stmt.limit(limit)
444 stmt = stmt.options(defer(File.data))
445 result = await db.execute(stmt)
446 rows = result.all()
448 items = []
449 for file, user, knowledge in rows:
450 items.append(
451 FileUserResponse(
452 id=file.id,
453 user_id=file.user_id,
454 hash=file.hash,
455 filename=file.filename,
456 meta=file.meta,
457 created_at=file.created_at,
458 updated_at=file.updated_at,
459 user=(UserResponse(**UserModel.model_validate(user).model_dump()) if user else None),
460 collection=(await self._to_knowledge_model(knowledge, db=db)).model_dump(),
461 )
462 )
464 return KnowledgeFileListResponse(items=items, total=total)
466 except Exception as e:
467 print('search_knowledge_files error:', e)
468 return KnowledgeFileListResponse(items=[], total=0)
470 async def check_access_by_user_id(
471 self,
472 id,
473 user_id,
474 permission='write',
475 db: Optional[AsyncSession] = None,
476 user_group_ids: set[str] | None = None,
477 ) -> bool:
478 knowledge = await self.get_knowledge_by_id(id, db=db)
479 if not knowledge:
480 return False
481 if knowledge.user_id == user_id:
482 return True
483 if user_group_ids is None:
484 user_groups = await Groups.get_groups_by_member_id(user_id, db=db)
485 user_group_ids = {group.id for group in user_groups}
486 return await AccessGrants.has_access(
487 user_id=user_id,
488 resource_type='knowledge',
489 resource_id=knowledge.id,
490 permission=permission,
491 user_group_ids=user_group_ids,
492 db=db,
493 )
495 async def get_knowledge_by_id(self, id: str, db: Optional[AsyncSession] = None) -> Optional[KnowledgeModel]:
496 try:
497 async with get_async_db_context(db) as db:
498 result = await db.execute(select(Knowledge).filter_by(id=id))
499 knowledge = result.scalars().first()
500 return await self._to_knowledge_model(knowledge, db=db) if knowledge else None
501 except Exception:
502 return None
504 async def get_knowledges_by_file_id(self, file_id: str, db: Optional[AsyncSession] = None) -> list[KnowledgeModel]:
505 try:
506 async with get_async_db_context(db) as db:
507 result = await db.execute(
508 select(Knowledge)
509 .join(KnowledgeFile, Knowledge.id == KnowledgeFile.knowledge_id)
510 .filter(KnowledgeFile.file_id == file_id)
511 )
512 knowledges = result.scalars().all()
513 knowledge_ids = [k.id for k in knowledges]
514 grants_map = await AccessGrants.get_grants_by_resources('knowledge', knowledge_ids, db=db)
515 return [
516 await self._to_knowledge_model(
517 knowledge,
518 access_grants=grants_map.get(knowledge.id, []),
519 db=db,
520 )
521 for knowledge in knowledges
522 ]
523 except Exception:
524 return []
526 async def search_files_by_id(
527 self,
528 knowledge_id: str,
529 user_id: str,
530 filter: dict,
531 skip: int = 0,
532 limit: int = 30,
533 db: Optional[AsyncSession] = None,
534 ) -> KnowledgeFileListResponse:
535 try:
536 async with get_async_db_context(db) as db:
537 stmt = (
538 select(File, User)
539 .join(KnowledgeFile, File.id == KnowledgeFile.file_id)
540 .outerjoin(User, User.id == KnowledgeFile.user_id)
541 .filter(KnowledgeFile.knowledge_id == knowledge_id)
542 )
544 # Filter by directory_id (None = root level)
545 directory_id = filter.get('directory_id') if filter else None
546 if directory_id:
547 stmt = stmt.filter(KnowledgeFile.directory_id == directory_id)
548 elif filter and 'directory_id' in filter:
549 # Explicit None = root level only
550 stmt = stmt.filter(KnowledgeFile.directory_id.is_(None))
552 # Default sort: updated_at descending
553 primary_sort = File.updated_at.desc()
555 if filter:
556 query_key = filter.get('query')
557 if query_key:
558 if filter.get('include_content'):
559 # Use ->> (as_string) instead of CAST(-> AS TEXT)
560 # to avoid PostgreSQL memory allocation failures on
561 # large content (#24670).
562 content_text = File.data['content'].as_string()
563 content_text = func.substr(content_text, 1, RAG_FILE_CONTENT_SEARCH_MAX_CHARS)
564 stmt = stmt.filter(
565 or_(
566 File.filename.ilike(f'%{query_key}%'),
567 content_text.ilike(f'%{query_key}%'),
568 )
569 )
570 else:
571 stmt = stmt.filter(File.filename.ilike(f'%{query_key}%'))
573 view_option = filter.get('view_option')
574 if view_option == 'created':
575 stmt = stmt.filter(KnowledgeFile.user_id == user_id)
576 elif view_option == 'shared':
577 stmt = stmt.filter(KnowledgeFile.user_id != user_id)
579 order_by = filter.get('order_by')
580 direction = filter.get('direction')
581 is_asc = direction == 'asc'
583 if order_by == 'name':
584 primary_sort = File.filename.asc() if is_asc else File.filename.desc()
585 elif order_by == 'created_at':
586 primary_sort = File.created_at.asc() if is_asc else File.created_at.desc()
587 elif order_by == 'updated_at':
588 primary_sort = File.updated_at.asc() if is_asc else File.updated_at.desc()
590 # Apply sort with secondary key for deterministic pagination
591 stmt = stmt.order_by(primary_sort, File.id.asc())
593 # Count BEFORE pagination
594 count_result = await db.execute(select(func.count()).select_from(stmt.subquery()))
595 total = count_result.scalar()
597 if skip:
598 stmt = stmt.offset(skip)
599 if limit:
600 stmt = stmt.limit(limit)
602 stmt = stmt.options(defer(File.data))
603 result = await db.execute(stmt)
604 items = result.all()
606 files = [
607 FileUserResponse(
608 id=file.id,
609 user_id=file.user_id,
610 hash=file.hash,
611 filename=file.filename,
612 meta=file.meta,
613 created_at=file.created_at,
614 updated_at=file.updated_at,
615 user=(UserResponse(**UserModel.model_validate(user).model_dump()) if user else None),
616 )
617 for file, user in items
618 ]
620 return KnowledgeFileListResponse(
621 items=files,
622 directories=await self.get_directories(
623 knowledge_id,
624 parent_id=filter.get('directory_id') if filter else None,
625 db=db,
626 ),
627 breadcrumbs=await self.get_directory_breadcrumbs(
628 knowledge_id,
629 filter.get('directory_id') if filter else None,
630 db=db,
631 ),
632 total=total,
633 )
634 except Exception as e:
635 print(e)
636 return KnowledgeFileListResponse(items=[], total=0)
638 async def get_files_by_id(self, knowledge_id: str, db: Optional[AsyncSession] = None) -> list[FileModel]:
639 try:
640 async with get_async_db_context(db) as db:
641 result = await db.execute(
642 select(File)
643 .join(KnowledgeFile, File.id == KnowledgeFile.file_id)
644 .filter(KnowledgeFile.knowledge_id == knowledge_id)
645 )
646 files = result.scalars().all()
647 return [FileModel.model_validate(file) for file in files]
648 except Exception:
649 return []
651 async def get_file_metadatas_by_id(
652 self, knowledge_id: str, db: Optional[AsyncSession] = None
653 ) -> list[FileMetadataResponse]:
654 """Column-only listing: File.data holds each file's full extracted
655 text, which metadata views must never load."""
656 try:
657 async with get_async_db_context(db) as db:
658 result = await db.execute(
659 select(File.id, File.hash, File.meta, File.created_at, File.updated_at)
660 .join(KnowledgeFile, File.id == KnowledgeFile.file_id)
661 .filter(KnowledgeFile.knowledge_id == knowledge_id)
662 )
663 return [
664 FileMetadataResponse(
665 id=row.id,
666 hash=row.hash,
667 meta=row.meta,
668 created_at=row.created_at,
669 updated_at=row.updated_at,
670 )
671 for row in result.all()
672 ]
673 except Exception:
674 return []
676 async def add_file_to_knowledge_by_id(
677 self,
678 knowledge_id: str,
679 file_id: str,
680 user_id: str,
681 directory_id: Optional[str] = None,
682 db: Optional[AsyncSession] = None,
683 ) -> Optional[KnowledgeFileModel]:
684 async with get_async_db_context(db) as db:
685 knowledge_file = KnowledgeFileModel(
686 **{
687 'id': str(uuid.uuid4()),
688 'knowledge_id': knowledge_id,
689 'file_id': file_id,
690 'directory_id': directory_id,
691 'user_id': user_id,
692 'created_at': int(time.time()),
693 'updated_at': int(time.time()),
694 }
695 )
697 try:
698 result = KnowledgeFile(**knowledge_file.model_dump())
699 db.add(result)
700 await db.commit()
701 await db.refresh(result)
702 if result:
703 return KnowledgeFileModel.model_validate(result)
704 else:
705 return None
706 except Exception:
707 return None
709 async def has_file(self, knowledge_id: str, file_id: str, db: Optional[AsyncSession] = None) -> bool:
710 """Check whether a file belongs to a knowledge base."""
711 try:
712 async with get_async_db_context(db) as db:
713 result = await db.execute(
714 select(KnowledgeFile).filter_by(knowledge_id=knowledge_id, file_id=file_id).limit(1)
715 )
716 return result.scalars().first() is not None
717 except Exception:
718 return False
720 async def remove_file_from_knowledge_by_id(
721 self, knowledge_id: str, file_id: str, db: Optional[AsyncSession] = None
722 ) -> bool:
723 try:
724 async with get_async_db_context(db) as db:
725 await db.execute(delete(KnowledgeFile).filter_by(knowledge_id=knowledge_id, file_id=file_id))
726 await db.commit()
727 return True
728 except Exception:
729 return False
731 async def reset_knowledge_by_id(
732 self, id: str, include_directories: bool = True, db: Optional[AsyncSession] = None
733 ) -> Optional[KnowledgeModel]:
734 try:
735 async with get_async_db_context(db) as db:
736 # Delete all knowledge_file entries for this knowledge_id
737 await db.execute(delete(KnowledgeFile).filter_by(knowledge_id=id))
739 # Delete all directories if requested
740 if include_directories:
741 await db.execute(delete(KnowledgeDirectory).filter_by(knowledge_id=id))
743 await db.commit()
745 # Update the knowledge entry's updated_at timestamp
746 await db.execute(update(Knowledge).filter_by(id=id).values(updated_at=int(time.time())))
747 await db.commit()
749 return await self.get_knowledge_by_id(id=id, db=db)
750 except Exception as e:
751 log.exception(e)
752 return None
754 async def update_knowledge_by_id(
755 self,
756 id: str,
757 form_data: KnowledgeForm,
758 overwrite: bool = False,
759 db: Optional[AsyncSession] = None,
760 ) -> Optional[KnowledgeModel]:
761 try:
762 async with get_async_db_context(db) as db:
763 await db.execute(
764 update(Knowledge)
765 .filter_by(id=id)
766 .values(
767 **form_data.model_dump(exclude={'access_grants'}),
768 updated_at=int(time.time()),
769 )
770 )
771 await db.commit()
772 if form_data.access_grants is not None:
773 await AccessGrants.set_access_grants('knowledge', id, form_data.access_grants, db=db)
774 return await self.get_knowledge_by_id(id=id, db=db)
775 except Exception as e:
776 log.exception(e)
777 return None
779 async def update_knowledge_meta_by_id(
780 self, id: str, meta: dict, db: Optional[AsyncSession] = None
781 ) -> Optional[KnowledgeModel]:
782 try:
783 async with get_async_db_context(db) as db:
784 await db.execute(
785 update(Knowledge)
786 .filter_by(id=id)
787 .values(
788 meta=meta,
789 updated_at=int(time.time()),
790 )
791 )
792 await db.commit()
793 return await self.get_knowledge_by_id(id=id, db=db)
794 except Exception as e:
795 log.exception(e)
796 return None
798 async def delete_knowledge_by_id(self, id: str, db: Optional[AsyncSession] = None) -> bool:
799 try:
800 async with get_async_db_context(db) as db:
801 await AccessGrants.revoke_all_access('knowledge', id, db=db)
802 await db.execute(delete(Knowledge).filter_by(id=id))
803 await db.commit()
804 return True
805 except Exception:
806 return False
808 async def delete_all_knowledge(self, db: Optional[AsyncSession] = None) -> bool:
809 async with get_async_db_context(db) as db:
810 try:
811 result = await db.execute(select(Knowledge.id))
812 knowledge_ids = [row[0] for row in result.all()]
813 for knowledge_id in knowledge_ids:
814 await AccessGrants.revoke_all_access('knowledge', knowledge_id, db=db)
815 await db.execute(delete(Knowledge))
816 await db.commit()
818 return True
819 except Exception:
820 return False
822 # ── Directory CRUD ────────────────────────────────────────────────
824 async def create_directory(
825 self,
826 knowledge_id: str,
827 name: str,
828 user_id: str,
829 parent_id: Optional[str] = None,
830 db: Optional[AsyncSession] = None,
831 ) -> Optional[KnowledgeDirectoryModel]:
832 async with get_async_db_context(db) as db:
833 try:
834 now = int(time.time())
835 directory = KnowledgeDirectory(
836 id=str(uuid.uuid4()),
837 knowledge_id=knowledge_id,
838 parent_id=parent_id,
839 name=name,
840 user_id=user_id,
841 created_at=now,
842 updated_at=now,
843 )
844 db.add(directory)
845 await db.commit()
846 await db.refresh(directory)
847 return KnowledgeDirectoryModel.model_validate(directory)
848 except Exception as e:
849 log.exception(e)
850 return None
852 async def get_directories(
853 self,
854 knowledge_id: str,
855 parent_id: Optional[str] = None,
856 db: Optional[AsyncSession] = None,
857 ) -> list[KnowledgeDirectoryModel]:
858 """List directories at a given level (parent_id=None for root)."""
859 async with get_async_db_context(db) as db:
860 stmt = select(KnowledgeDirectory).filter(KnowledgeDirectory.knowledge_id == knowledge_id)
861 if parent_id:
862 stmt = stmt.filter(KnowledgeDirectory.parent_id == parent_id)
863 else:
864 stmt = stmt.filter(KnowledgeDirectory.parent_id.is_(None))
866 stmt = stmt.order_by(KnowledgeDirectory.name.asc())
867 result = await db.execute(stmt)
868 return [KnowledgeDirectoryModel.model_validate(d) for d in result.scalars().all()]
870 async def get_all_directories(
871 self,
872 knowledge_id: str,
873 db: Optional[AsyncSession] = None,
874 ) -> list[KnowledgeDirectoryModel]:
875 """Get ALL directories for a KB (no parent filter). Used for tree building."""
876 async with get_async_db_context(db) as db:
877 stmt = (
878 select(KnowledgeDirectory)
879 .filter(KnowledgeDirectory.knowledge_id == knowledge_id)
880 .order_by(KnowledgeDirectory.name.asc())
881 )
882 result = await db.execute(stmt)
883 return [KnowledgeDirectoryModel.model_validate(d) for d in result.scalars().all()]
885 async def get_files_with_directory_ids(
886 self,
887 knowledge_id: str,
888 db: Optional[AsyncSession] = None,
889 ) -> list[tuple[FileModel, Optional[str]]]:
890 """Get all files in a KB with their directory_id from KnowledgeFile."""
891 try:
892 async with get_async_db_context(db) as db:
893 result = await db.execute(
894 select(File, KnowledgeFile.directory_id)
895 .join(KnowledgeFile, File.id == KnowledgeFile.file_id)
896 .filter(KnowledgeFile.knowledge_id == knowledge_id)
897 )
898 return [(FileModel.model_validate(file), dir_id) for file, dir_id in result.all()]
899 except Exception:
900 return []
902 async def get_directory_by_id(
903 self, directory_id: str, db: Optional[AsyncSession] = None
904 ) -> Optional[KnowledgeDirectoryModel]:
905 async with get_async_db_context(db) as db:
906 result = await db.execute(select(KnowledgeDirectory).filter_by(id=directory_id))
907 directory = result.scalars().first()
908 return KnowledgeDirectoryModel.model_validate(directory) if directory else None
910 async def get_directory_breadcrumbs(
911 self,
912 knowledge_id: str,
913 directory_id: Optional[str],
914 db: Optional[AsyncSession] = None,
915 ) -> list[KnowledgeDirectoryModel]:
916 """Walk up the parent chain to build breadcrumbs (root first)."""
917 if not directory_id:
918 return []
920 async with get_async_db_context(db) as db:
921 breadcrumbs = []
922 current_id = directory_id
923 seen = set()
925 while current_id and current_id not in seen:
926 seen.add(current_id)
927 # Scoped by knowledge base so a caller-supplied id cannot walk another one's tree.
928 result = await db.execute(
929 select(KnowledgeDirectory).filter_by(id=current_id, knowledge_id=knowledge_id)
930 )
931 directory = result.scalars().first()
932 if not directory:
933 break
934 breadcrumbs.append(KnowledgeDirectoryModel.model_validate(directory))
935 current_id = directory.parent_id
937 breadcrumbs.reverse() # root first
938 return breadcrumbs
940 async def rename_directory(
941 self,
942 directory_id: str,
943 name: str,
944 db: Optional[AsyncSession] = None,
945 ) -> Optional[KnowledgeDirectoryModel]:
946 async with get_async_db_context(db) as db:
947 try:
948 await db.execute(
949 update(KnowledgeDirectory).filter_by(id=directory_id).values(name=name, updated_at=int(time.time()))
950 )
951 await db.commit()
952 return await self.get_directory_by_id(directory_id, db=db)
953 except Exception as e:
954 log.exception(e)
955 return None
957 async def move_directory(
958 self,
959 directory_id: str,
960 new_parent_id: Optional[str],
961 db: Optional[AsyncSession] = None,
962 ) -> Optional[KnowledgeDirectoryModel]:
963 """Move a directory to a new parent, with cycle detection."""
964 async with get_async_db_context(db) as db:
965 try:
966 # Cycle detection: walk up from new_parent_id to ensure
967 # we don't encounter directory_id
968 if new_parent_id:
969 current = new_parent_id
970 seen = set()
971 while current and current not in seen:
972 if current == directory_id:
973 return None # Would create a cycle
974 seen.add(current)
975 result = await db.execute(select(KnowledgeDirectory.parent_id).filter_by(id=current))
976 row = result.first()
977 current = row[0] if row else None
979 await db.execute(
980 update(KnowledgeDirectory)
981 .filter_by(id=directory_id)
982 .values(parent_id=new_parent_id, updated_at=int(time.time()))
983 )
984 await db.commit()
985 return await self.get_directory_by_id(directory_id, db=db)
986 except Exception as e:
987 log.exception(e)
988 return None
990 async def update_directory(
991 self,
992 directory_id: str,
993 name: Optional[str] = None,
994 parent_id: Optional[str] = '__unset__',
995 db: Optional[AsyncSession] = None,
996 ) -> Optional[KnowledgeDirectoryModel]:
997 """Update directory name and/or parent. Pass parent_id=None to move to root."""
998 # Handle move if parent_id is being changed
999 if parent_id != '__unset__':
1000 result = await self.move_directory(directory_id, parent_id, db=db)
1001 if result is None:
1002 return None # Cycle detected or error
1004 if name is not None:
1005 return await self.rename_directory(directory_id, name, db=db)
1007 return await self.get_directory_by_id(directory_id, db=db)
1009 async def delete_directory(
1010 self,
1011 directory_id: str,
1012 move_files_to_parent: bool = True,
1013 db: Optional[AsyncSession] = None,
1014 ) -> bool:
1015 """
1016 Delete a directory.
1017 - If move_files_to_parent=True: files move to parent dir (or root)
1018 - If move_files_to_parent=False: files are also deleted
1019 """
1020 async with get_async_db_context(db) as db:
1021 try:
1022 # Get the directory to find its parent
1023 result = await db.execute(select(KnowledgeDirectory).filter_by(id=directory_id))
1024 directory = result.scalars().first()
1025 if not directory:
1026 return False
1028 parent_id = directory.parent_id
1030 if move_files_to_parent:
1031 # Move files in this directory to its parent (or root)
1032 await db.execute(
1033 update(KnowledgeFile).filter_by(directory_id=directory_id).values(directory_id=parent_id)
1034 )
1035 # Recursively move files from all subdirectories too
1036 await self._move_files_from_subtree(directory_id, parent_id, db=db)
1037 else:
1038 # Delete files in this directory and all subdirectories
1039 await self._delete_files_in_subtree(directory_id, db=db)
1041 # CASCADE on parent_id will handle deleting subdirectories
1042 await db.execute(delete(KnowledgeDirectory).filter_by(id=directory_id))
1043 await db.commit()
1044 return True
1045 except Exception as e:
1046 log.exception(e)
1047 return False
1049 async def _move_files_from_subtree(
1050 self,
1051 directory_id: str,
1052 target_directory_id: Optional[str],
1053 db: AsyncSession,
1054 ) -> None:
1055 """Recursively move all files from a directory subtree to the target."""
1056 result = await db.execute(select(KnowledgeDirectory.id).filter_by(parent_id=directory_id))
1057 child_ids = [row[0] for row in result.all()]
1059 for child_id in child_ids:
1060 await db.execute(
1061 update(KnowledgeFile).filter_by(directory_id=child_id).values(directory_id=target_directory_id)
1062 )
1063 await self._move_files_from_subtree(child_id, target_directory_id, db=db)
1065 async def _delete_files_in_subtree(
1066 self,
1067 directory_id: str,
1068 db: AsyncSession,
1069 ) -> None:
1070 """Recursively delete all files from a directory subtree."""
1071 await db.execute(delete(KnowledgeFile).filter_by(directory_id=directory_id))
1072 result = await db.execute(select(KnowledgeDirectory.id).filter_by(parent_id=directory_id))
1073 child_ids = [row[0] for row in result.all()]
1074 for child_id in child_ids:
1075 await self._delete_files_in_subtree(child_id, db=db)
1077 async def get_files_by_id_and_directory_id(
1078 self,
1079 knowledge_id: str,
1080 directory_id: str,
1081 db: Optional[AsyncSession] = None,
1082 ) -> list[FileModel]:
1083 """Get all files in a directory and its subdirectories."""
1084 async with get_async_db_context(db) as db:
1085 directory_ids = [directory_id]
1086 for parent_id in directory_ids:
1087 result = await db.execute(select(KnowledgeDirectory.id).filter_by(parent_id=parent_id))
1088 directory_ids.extend(result.scalars().all())
1089 result = await db.execute(
1090 select(File)
1091 .join(KnowledgeFile, File.id == KnowledgeFile.file_id)
1092 .filter(KnowledgeFile.knowledge_id == knowledge_id)
1093 .filter(KnowledgeFile.directory_id.in_(directory_ids))
1094 )
1095 return [FileModel.model_validate(file) for file in result.scalars().all()]
1097 async def move_file_to_directory(
1098 self,
1099 knowledge_id: str,
1100 file_id: str,
1101 directory_id: Optional[str] = None,
1102 db: Optional[AsyncSession] = None,
1103 ) -> bool:
1104 """Move a file to a different directory within the same KB."""
1105 async with get_async_db_context(db) as db:
1106 try:
1107 await db.execute(
1108 update(KnowledgeFile)
1109 .filter_by(knowledge_id=knowledge_id, file_id=file_id)
1110 .values(directory_id=directory_id, updated_at=int(time.time()))
1111 )
1112 await db.commit()
1113 return True
1114 except Exception as e:
1115 log.exception(e)
1116 return False
1119Knowledges = KnowledgeTable()