Coverage for open_webui/models/chats.py: 56%
1409 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 05:07 +0000
1"""Chat models, forms, and database operations."""
3from __future__ import annotations
5import logging
6import re
7import time
8import uuid
9from typing import Any, Literal
11# local imports
12from open_webui.env import ENABLE_ADMIN_CHAT_ACCESS
13from open_webui.internal.db import Base, JSONField, get_async_db_context
14from open_webui.models.access_grants import AccessGrants
15from open_webui.models.automations import AutomationRun
16from open_webui.models.chat_messages import ChatMessage, ChatMessages
17from open_webui.models.folders import Folders
18from open_webui.models.tags import Tag, TagModel, Tags
19from open_webui.utils.misc import get_output_text, sanitize_data_for_db, sanitize_text_for_db
20from pydantic import BaseModel, ConfigDict, field_validator
21from sqlalchemy import (
22 JSON,
23 BigInteger,
24 Boolean,
25 Column,
26 ForeignKey,
27 Index,
28 String,
29 Text,
30 UniqueConstraint,
31 and_,
32 delete,
33 exists,
34 func,
35 or_,
36 select,
37 text,
38 update,
39)
40from sqlalchemy.ext.asyncio import AsyncSession
41from sqlalchemy.orm.attributes import flag_modified
42from sqlalchemy.sql import case, exists
43from sqlalchemy.sql.expression import bindparam
45log = logging.getLogger(__name__)
46ACTIVE_CHAT_GAP_SECONDS = 30 * 60
47CHAT_SEARCH_FILTER_PREFIXES = ('tag:', 'folder:', 'pinned:', 'archived:', 'shared:')
50def chat_search_content_query(text: str) -> str:
51 words = sanitize_text_for_db(text).lower().strip().split()
52 return ' '.join(word for word in words if not word.startswith(CHAT_SEARCH_FILTER_PREFIXES)).strip()
55def chat_search_terms(text: str) -> list[str]:
56 return list(dict.fromkeys(re.findall(r'[a-z0-9]+', text.lower())))
59def chat_search_message_content_match_sql(dialect_name: str, key: str) -> str:
60 if dialect_name == 'sqlite': 60 ↛ 76line 60 didn't jump to line 76 because the condition on line 60 was always true
61 return f"""
62 (
63 EXISTS (
64 SELECT 1
65 FROM json_each(Chat.chat, '$.history.messages') AS history_message
66 WHERE LOWER(history_message.value->>'content') LIKE '%' || :{key} || '%'
67 )
68 OR EXISTS (
69 SELECT 1
70 FROM json_each(Chat.chat, '$.messages') AS legacy_message
71 WHERE LOWER(legacy_message.value->>'content') LIKE '%' || :{key} || '%'
72 )
73 )
74 """
76 if dialect_name == 'postgresql':
77 return f"""
78 (
79 EXISTS (
80 SELECT 1
81 FROM chat_message AS message
82 WHERE message.chat_id = Chat.id
83 AND message.user_id = Chat.user_id
84 AND json_typeof(message.content) = 'string'
85 AND LOWER(message.content #>> '{{}}') LIKE '%' || :{key} || '%'
86 )
87 OR EXISTS (
88 SELECT 1
89 FROM json_each(Chat.chat#>'{{history,messages}}') AS history_message
90 WHERE json_typeof(history_message.value->'content') = 'string'
91 AND LOWER(history_message.value->>'content') LIKE '%' || :{key} || '%'
92 )
93 OR EXISTS (
94 SELECT 1
95 FROM json_array_elements(Chat.chat->'messages') AS legacy_message
96 WHERE json_typeof(legacy_message->'content') = 'string'
97 AND LOWER(legacy_message->>'content') LIKE '%' || :{key} || '%'
98 )
99 )
100 """
102 raise NotImplementedError(f'Unsupported dialect: {dialect_name}')
105def chat_list_order(sort_by: str = 'updated_at', sort_dir: str = 'desc', user_id: str | None = None):
106 if sort_by != 'unread_updated_at':
107 sort_column = Chat.title if sort_by == 'title' else Chat.updated_at
108 order_clause = sort_column.asc() if sort_dir == 'asc' else sort_column.desc()
109 return order_clause, Chat.id
111 unfinished_assistant = (
112 select(ChatMessage.id)
113 .where(ChatMessage.chat_id == Chat.id)
114 .where(ChatMessage.role == 'assistant')
115 .where(ChatMessage.done.is_(False))
116 .exists()
117 )
118 conditions = [Chat.updated_at > func.coalesce(Chat.last_read_at, 0), ~unfinished_assistant]
119 if user_id is not None:
120 conditions.append(Chat.user_id == user_id)
122 unread = case(
123 (and_(*conditions), 1),
124 else_=0,
125 )
126 return unread.desc(), Chat.updated_at.desc(), Chat.id
129class Chat(Base): # database table mapping for chat entity
130 __tablename__ = 'chat'
132 id = Column(String, primary_key=True, unique=True)
133 user_id = Column(String, index=True) # owner user id
134 title = Column(Text) # user-visible conversation title
135 chat = Column(JSON)
137 created_at = Column(BigInteger, index=True) # conversation creation timestamp
138 updated_at = Column(BigInteger, index=True) # conversation modification timestamp
140 share_id = Column(Text, unique=True, nullable=True) # public share link token
141 archived = Column(Boolean, default=False) # hidden from main chat list
142 pinned = Column(Boolean, default=False, nullable=True)
144 meta = Column(JSON, server_default='{}')
145 variables = Column(JSON, nullable=True)
146 folder_id = Column(Text, nullable=True)
148 tasks = Column(JSON, nullable=True)
149 summary = Column(Text, nullable=True)
150 current_message_id = Column(Text, nullable=True)
152 last_read_at = Column(BigInteger, nullable=True)
153 timer_at = Column(BigInteger, nullable=True) # ns due time, set only while a timer chat waits to be claimed
155 __table_args__ = (
156 # Performance indexes for common queries
157 Index('folder_id_idx', 'folder_id'),
158 Index('user_id_pinned_idx', 'user_id', 'pinned'),
159 Index('user_id_archived_idx', 'user_id', 'archived'),
160 Index('updated_at_user_id_idx', 'updated_at', 'user_id'),
161 Index('folder_id_user_id_idx', 'folder_id', 'user_id'),
162 Index('user_id_updated_at_id_idx', 'user_id', updated_at.desc(), 'id'),
163 Index(
164 'timer_at_idx',
165 'timer_at',
166 sqlite_where=text('timer_at IS NOT NULL'),
167 postgresql_where=text('timer_at IS NOT NULL'),
168 ),
169 # timer_at key column turns the IS NOT NULL into a seek, so this beats the plain user_id indexes
170 Index(
171 'user_id_timer_at_idx',
172 'user_id',
173 'timer_at',
174 sqlite_where=text('timer_at IS NOT NULL'),
175 postgresql_where=text('timer_at IS NOT NULL'),
176 ),
177 # covering index: lets SQLite serve count_unread_by_folder_ids without reading chat rows
178 Index('user_id_folder_unread_idx', 'user_id', 'folder_id', 'archived', 'updated_at', 'last_read_at', 'id'),
179 )
182def is_internal_chat(meta: dict | None) -> bool:
183 return bool(meta and meta.get('internal') is True)
186class ChatModel(BaseModel):
187 model_config = ConfigDict(from_attributes=True) # allows ORM model binding
188 id: str
189 user_id: str
190 title: str
191 chat: dict
193 created_at: int # timestamp in epoch
194 updated_at: int # timestamp in epoch
196 share_id: str | None = None
197 archived: bool = False
198 pinned: bool | None = False
200 meta: dict = {}
201 variables: dict = {}
202 folder_id: str | None = None
204 tasks: list | None = None
205 summary: str | None = None
206 current_message_id: str | None = None
208 last_read_at: int | None = None
209 timer_at: int | None = None
211 @field_validator('variables', mode='before')
212 @classmethod
213 def normalize_variables(cls, value):
214 return value if isinstance(value, dict) else {}
217class ChatFile(Base):
218 __tablename__ = 'chat_file'
220 id = Column(Text, unique=True, primary_key=True)
221 user_id = Column(Text, nullable=False)
223 chat_id = Column(Text, ForeignKey('chat.id', ondelete='CASCADE'), nullable=False)
224 message_id = Column(Text, nullable=True)
225 file_id = Column(Text, ForeignKey('file.id', ondelete='CASCADE'), nullable=False)
227 created_at = Column(BigInteger, nullable=False)
228 updated_at = Column(BigInteger, nullable=False)
230 __table_args__ = (UniqueConstraint('chat_id', 'file_id', name='uq_chat_file_chat_file'),)
233class ChatFileModel(BaseModel):
234 id: str
235 user_id: str
237 chat_id: str
238 message_id: str | None = None
239 file_id: str
241 created_at: int
242 updated_at: int
244 model_config = ConfigDict(from_attributes=True)
247####################
248# Forms
249####################
252class ChatForm(BaseModel):
253 chat: dict
254 variables: dict | None = None
255 folder_id: str | None = None
258class ChatImportForm(ChatForm):
259 meta: dict | None = {}
260 pinned: bool | None = False
261 archived: bool | None = False
262 current_message_id: str | None = None
263 created_at: int | None = None
264 updated_at: int | None = None
267class ChatsImportForm(BaseModel):
268 chats: list[ChatImportForm]
271class ChatTitleForm(BaseModel):
272 title: str
275class ChatResponse(BaseModel):
276 id: str
277 user_id: str
278 title: str
279 chat: dict
280 updated_at: int # timestamp in epoch
281 created_at: int # timestamp in epoch
282 share_id: str | None = None # id of the chat to be shared
283 archived: bool
284 pinned: bool | None = False
285 meta: dict = {}
286 variables: dict = {}
287 folder_id: str | None = None
289 tasks: list | None = None
290 summary: str | None = None
291 current_message_id: str | None = None
292 context_usage: dict | None = None
294 @field_validator('variables', mode='before')
295 @classmethod
296 def normalize_variables(cls, value):
297 return value if isinstance(value, dict) else {}
300class ChatTitleIdResponse(BaseModel):
301 id: str
302 title: str
303 updated_at: int
304 created_at: int
305 last_read_at: int | None = None
306 snippet: str | None = None
307 active: bool = False
308 archived: bool = False
311class SharedChatResponse(BaseModel):
312 id: str
313 title: str
314 share_id: str | None = None
315 updated_at: int
316 created_at: int
319class ChatListResponse(BaseModel):
320 items: list[ChatModel]
321 total: int
324class ChatUsageStatsResponse(BaseModel):
325 id: str # chat id
327 models: dict = {} # models used in the chat with their usage counts
328 message_count: int # number of messages in the chat
330 history_models: dict = {} # models used in the chat history with their usage counts
331 history_message_count: int # number of messages in the chat history
332 history_user_message_count: int # number of user messages in the chat history
333 history_assistant_message_count: int # number of assistant messages in the chat history
335 average_response_time: float # average response time of assistant messages in seconds
336 average_user_message_content_length: float # average length of user message contents
337 average_assistant_message_content_length: float # average length of assistant message contents
339 tags: list[str] = [] # tags associated with the chat
341 last_message_at: int # timestamp of the last message
342 updated_at: int
343 created_at: int
345 model_config = ConfigDict(extra='allow')
348class ChatUsageStatsListResponse(BaseModel):
349 items: list[ChatUsageStatsResponse]
350 total: int
351 model_config = ConfigDict(extra='allow')
354class MessageStats(BaseModel):
355 id: str
356 role: str
357 model: str | None = None
358 content_length: int
359 token_count: int | None = None
360 timestamp: int | None = None
361 rating: int | None = None # Derived from message.annotation.rating
362 tags: list[str] | None = None # Derived from message.annotation.tags
365class ChatHistoryStats(BaseModel):
366 messages: dict[str, MessageStats]
367 currentId: str | None = None
370class ChatBody(BaseModel):
371 history: ChatHistoryStats
374class AggregateChatStats(BaseModel):
375 average_response_time: float
376 average_user_message_content_length: float
377 average_assistant_message_content_length: float
378 models: dict[str, int]
379 message_count: int
380 history_models: dict[str, int]
381 history_message_count: int
382 history_user_message_count: int
383 history_assistant_message_count: int
386class ChatStatsExport(BaseModel):
387 id: str
388 user_id: str
389 created_at: int
390 updated_at: int
391 tags: list[str] = []
392 stats: AggregateChatStats
393 chat: ChatBody
396class ChatTable:
397 def _clean_null_bytes(self, obj):
398 """Recursively remove null bytes from strings in dict/list structures."""
399 return sanitize_data_for_db(obj)
401 def get_current_message_id(self, chat: dict | None) -> str | None:
402 chat = chat or {}
403 history = chat.get('history') if isinstance(chat.get('history'), dict) else {}
404 current_id = history.get('currentId') or chat.get('currentId') or chat.get('branchPointMessageId')
405 if current_id:
406 return current_id
408 messages = chat.get('messages')
409 if isinstance(messages, list):
410 for message in reversed(messages): 410 ↛ 411line 410 didn't jump to line 411 because the loop on line 410 never started
411 if isinstance(message, dict) and message.get('id'):
412 return message['id']
414 return None
416 def _sanitize_chat_row(self, chat_item):
417 """
418 Clean a Chat SQLAlchemy model's title + chat JSON,
419 and return True if anything changed.
420 """
421 changed = False
423 # Clean title
424 if chat_item.title: 424 ↛ 431line 424 didn't jump to line 431 because the condition on line 424 was always true
425 cleaned = self._clean_null_bytes(chat_item.title)
426 if cleaned != chat_item.title: 426 ↛ 427line 426 didn't jump to line 427 because the condition on line 426 was never true
427 chat_item.title = cleaned
428 changed = True
430 # Clean JSON
431 if chat_item.chat:
432 cleaned = self._clean_null_bytes(chat_item.chat)
433 if cleaned != chat_item.chat: 433 ↛ 434line 433 didn't jump to line 434 because the condition on line 433 was never true
434 chat_item.chat = cleaned
435 changed = True
437 return changed
439 @staticmethod
440 def _last_descendant_id(messages: dict, message_id: str) -> str:
441 seen_ids = set()
442 while message_id in messages and message_id not in seen_ids:
443 seen_ids.add(message_id)
444 message = messages[message_id]
445 child_ids = message.get('childrenIds') if isinstance(message, dict) else []
446 child_ids = child_ids if isinstance(child_ids, list) else []
447 next_id = next((child_id for child_id in reversed(child_ids) if child_id in messages), None)
448 if not next_id:
449 break
450 message_id = next_id
451 return message_id
453 @staticmethod
454 def _add_child_id_to_parent(messages: dict, parent_id: str | None, child_id: str) -> bool:
455 parent = messages.get(parent_id) if parent_id else None
456 if not isinstance(parent, dict): 456 ↛ 459line 456 didn't jump to line 459 because the condition on line 456 was always true
457 return False
459 child_ids = parent.get('childrenIds')
460 if not isinstance(child_ids, list):
461 child_ids = []
462 parent['childrenIds'] = child_ids
464 if child_id in child_ids:
465 return False
467 child_ids.append(child_id)
468 return True
470 def _repair_chat_current_id(self, chat: dict) -> bool:
471 history = chat.get('history')
472 if not isinstance(history, dict):
473 return False
475 messages = history.get('messages')
476 if not isinstance(messages, dict): 476 ↛ 477line 476 didn't jump to line 477 because the condition on line 476 was never true
477 return False
479 changed = False
480 for message_id, message in messages.items():
481 if not isinstance(message, dict): 481 ↛ 482line 481 didn't jump to line 482 because the condition on line 481 was never true
482 continue
483 changed = self._add_child_id_to_parent(messages, message.get('parentId'), message_id) or changed
485 current_id = history.get('currentId')
486 current_message = messages.get(current_id)
487 output = []
488 if isinstance(current_message, dict):
489 output = current_message.get('output') or []
491 output_role = next(
492 (item.get('role') for item in output if isinstance(item, dict) and item.get('role')),
493 None,
494 )
495 current_is_bad_leaf = (
496 isinstance(current_message, dict)
497 and output_role == 'assistant'
498 and current_message.get('parentId') is None
499 and not current_message.get('timestamp')
500 and len(messages) > 1
501 )
502 if (
503 isinstance(current_message, dict)
504 and current_message.get('id')
505 and current_message.get('role')
506 and not current_is_bad_leaf
507 ):
508 if current_message.get('contextSummary') or current_message.get('context_summary'): 508 ↛ 509line 508 didn't jump to line 509 because the condition on line 508 was never true
509 last_descendant_id = self._last_descendant_id(messages, current_id)
510 if last_descendant_id != current_id:
511 history['currentId'] = last_descendant_id
512 return True
514 return changed
516 latest_leaf_id = None
517 latest_timestamp = -1
518 for message_id, message in messages.items(): 518 ↛ 519line 518 didn't jump to line 519 because the loop on line 518 never started
519 if not isinstance(message, dict) or not message.get('role'):
520 continue
522 children_ids = message.get('childrenIds') if isinstance(message.get('childrenIds'), list) else []
523 timestamp = message.get('timestamp') or 0
524 if len(children_ids) == 0 and timestamp > latest_timestamp:
525 latest_leaf_id = message_id
526 latest_timestamp = timestamp
528 if not latest_leaf_id or latest_leaf_id == current_id: 528 ↛ 531line 528 didn't jump to line 531 because the condition on line 528 was always true
529 return changed
531 history['currentId'] = latest_leaf_id
532 return True
534 async def insert_new_chat(
535 self,
536 id: str,
537 user_id: str,
538 form_data: ChatForm,
539 db: AsyncSession | None = None,
540 *,
541 internal_meta: dict | None = None,
542 timer_at: int | None = None,
543 ) -> ChatModel | None:
544 async with get_async_db_context(db) as session:
545 chat = ChatModel(
546 **{
547 'id': id,
548 'user_id': user_id,
549 'title': self._clean_null_bytes(
550 form_data.chat['title'] if 'title' in form_data.chat else 'New Chat'
551 ),
552 'chat': self._clean_null_bytes(form_data.chat),
553 'folder_id': form_data.folder_id,
554 'meta': internal_meta or {},
555 'timer_at': timer_at,
556 'variables': form_data.variables or {},
557 'current_message_id': self.get_current_message_id(form_data.chat),
558 'created_at': int(time.time()),
559 'updated_at': int(time.time()),
560 'last_read_at': int(time.time()),
561 }
562 )
564 chat_item = Chat(**chat.model_dump())
565 session.add(chat_item)
566 await session.commit()
568 # Dual-write initial messages to chat_message table
569 try:
570 history = form_data.chat.get('history') if isinstance(form_data.chat.get('history'), dict) else {}
571 messages = history.get('messages') if isinstance(history.get('messages'), dict) else {}
572 if not messages and isinstance(form_data.chat.get('messages'), list):
573 messages = {
574 message.get('id'): message
575 for message in form_data.chat['messages']
576 if isinstance(message, dict) and message.get('id')
577 }
578 for message_id, message in messages.items(): 578 ↛ 579line 578 didn't jump to line 579 because the loop on line 578 never started
579 if isinstance(message, dict) and message.get('role'):
580 await ChatMessages.upsert_message(
581 message_id=message_id,
582 chat_id=id,
583 user_id=user_id,
584 data=message,
585 )
586 except Exception as e:
587 log.warning(f'Failed to write initial messages to chat_message table: {e}')
589 return ChatModel.model_validate(chat_item) if chat_item else None
591 async def get_internal_chat_ids_by_parent_id(self, parent_chat_id: str, user_id: str) -> list[str]:
592 async with get_async_db_context() as session:
593 result = await session.execute(
594 select(Chat.id).where(
595 Chat.user_id == user_id,
596 Chat.meta['internal'].as_boolean().is_(True),
597 Chat.meta['parent_chat_id'].as_string() == parent_chat_id,
598 )
599 )
600 return list(result.scalars().all())
602 async def get_internal_chat_by_note_id(
603 self, note_id: str, user_id: str, db: AsyncSession | None = None
604 ) -> ChatModel | None:
605 async with get_async_db_context(db) as session:
606 result = await session.execute(
607 select(Chat)
608 .where(
609 Chat.user_id == user_id,
610 Chat.meta['internal'].as_boolean().is_(True),
611 Chat.meta['type'].as_string() == 'note',
612 Chat.meta['note_id'].as_string() == note_id,
613 )
614 .order_by(Chat.updated_at.desc(), Chat.created_at.desc())
615 )
616 chat = result.scalars().first()
617 return ChatModel.model_validate(chat) if chat else None
619 async def get_internal_chats_by_note_id(
620 self, note_id: str, user_id: str, db: AsyncSession | None = None
621 ) -> list[ChatModel]:
622 async with get_async_db_context(db) as session:
623 result = await session.execute(
624 select(Chat)
625 .where(
626 Chat.user_id == user_id,
627 Chat.meta['internal'].as_boolean().is_(True),
628 Chat.meta['type'].as_string() == 'note',
629 Chat.meta['note_id'].as_string() == note_id,
630 )
631 .order_by(Chat.updated_at.desc(), Chat.created_at.desc())
632 )
633 return [ChatModel.model_validate(chat) for chat in result.scalars().all()]
635 def _chat_import_form_to_chat_model(self, user_id: str, form_data: ChatImportForm) -> ChatModel:
636 id = str(uuid.uuid4())
637 chat = ChatModel(
638 **{
639 'id': id,
640 'user_id': user_id,
641 'title': self._clean_null_bytes(form_data.chat['title'] if 'title' in form_data.chat else 'New Chat'),
642 'chat': self._clean_null_bytes(form_data.chat),
643 'meta': form_data.meta,
644 'variables': form_data.variables or {},
645 'pinned': form_data.pinned,
646 'archived': form_data.archived,
647 'folder_id': form_data.folder_id,
648 'current_message_id': form_data.current_message_id or self.get_current_message_id(form_data.chat),
649 'created_at': (form_data.created_at if form_data.created_at else int(time.time())),
650 'updated_at': (form_data.updated_at if form_data.updated_at else int(time.time())),
651 }
652 )
653 return chat
655 async def import_chats(
656 self,
657 user_id: str,
658 chat_import_forms: list[ChatImportForm],
659 db: AsyncSession | None = None,
660 ) -> list[ChatModel]:
661 async with get_async_db_context(db) as session:
662 # Validate folder_id references — clear any that don't exist
663 folder_ids = {f.folder_id for f in chat_import_forms if f.folder_id}
664 existing = set()
665 for fid in folder_ids:
666 if await Folders.get_folder_by_id_and_user_id(fid, user_id, db=session): 666 ↛ 667line 666 didn't jump to line 667 because the condition on line 666 was never true
667 existing.add(fid)
669 cleared = 0
670 for form in chat_import_forms:
671 if form.folder_id and form.folder_id not in existing:
672 form.folder_id = None
673 cleared += 1
674 if cleared:
675 log.info('Import: cleared %d dangling folder_id(s) for user %s', cleared, user_id)
677 chats = []
679 for form_data in chat_import_forms:
680 chat = self._chat_import_form_to_chat_model(user_id, form_data)
681 chats.append(Chat(**chat.model_dump()))
683 session.add_all(chats)
684 await session.commit()
686 # Dual-write messages to chat_message table
687 for form_data, imported_chat in zip(chat_import_forms, chats): 687 ↛ 688line 687 didn't jump to line 688 because the loop on line 687 never started
688 history = form_data.chat.get('history') if isinstance(form_data.chat.get('history'), dict) else {}
689 messages = history.get('messages') if isinstance(history.get('messages'), dict) else {}
690 if not messages and isinstance(form_data.chat.get('messages'), list):
691 messages = {
692 message.get('id'): message
693 for message in form_data.chat['messages']
694 if isinstance(message, dict) and message.get('id')
695 }
696 for message_id, message in messages.items():
697 if isinstance(message, dict) and message.get('role'): 697 ↛ 696line 697 didn't jump to line 696 because the condition on line 697 was always true
698 try:
699 await ChatMessages.upsert_message(
700 message_id=message_id,
701 chat_id=imported_chat.id,
702 user_id=user_id,
703 data=message,
704 )
705 except Exception as e:
706 log.warning(
707 f'Failed to write imported message {message_id} for chat {imported_chat.id}: {e}'
708 )
710 return [ChatModel.model_validate(chat) for chat in chats]
712 async def update_chat_by_id(
713 self,
714 id: str,
715 chat: dict,
716 db: AsyncSession | None = None,
717 *,
718 touch: bool = True,
719 ) -> ChatModel | None:
720 """Patch top-level chat keys; history is merged so stale writers don't drop messages."""
721 try:
722 async with get_async_db_context(db) as session:
723 chat_item = await session.get(
724 Chat,
725 id,
726 populate_existing=True,
727 with_for_update=session.bind.dialect.name == 'postgresql',
728 )
729 if chat_item is None:
730 return None
732 stored = chat_item.chat or {}
733 updated = {**stored, **chat}
734 if 'history' in chat:
735 # The caller built its history from an earlier read; merge so messages saved since then survive.
736 updated['history'] = self.merge_history(stored.get('history'), chat['history'])
738 updated = self._clean_null_bytes(updated)
739 chat_item.chat = updated
740 chat_item.title = updated.get('title', 'New Chat')
741 if any(key in chat for key in ('history', 'messages', 'currentId', 'branchPointMessageId')): 741 ↛ anywhereline 741 didn't jump anywhere: it always raised an exception.
742 chat_item.current_message_id = self.get_current_message_id(updated)
744 if touch:
745 chat_item.updated_at = int(time.time())
747 await session.commit()
749 return ChatModel.model_validate(chat_item)
750 except Exception:
751 return
753 async def update_chat_variables_by_id(
754 self,
755 id: str,
756 variables: dict | None,
757 db: AsyncSession | None = None,
758 *,
759 touch: bool = True,
760 ) -> ChatModel | None:
761 try:
762 async with get_async_db_context(db) as session:
763 chat_item = await session.get(Chat, id)
764 if chat_item is None:
765 return None
767 chat_item.variables = variables if isinstance(variables, dict) else {}
768 if touch:
769 chat_item.updated_at = int(time.time())
771 await session.commit()
772 return ChatModel.model_validate(chat_item)
773 except Exception:
774 return None
776 async def update_chat_last_read_at_by_id(
777 self, id: str, user_id: str, db: AsyncSession | None = None
778 ) -> tuple[int, bool] | None:
779 try:
780 async with get_async_db_context(db) as session:
781 chat = await session.get(Chat, id)
782 if chat and chat.user_id == user_id:
783 last_read_at = int(time.time())
784 was_unread = chat.last_read_at is None or chat.updated_at > chat.last_read_at
785 chat.last_read_at = last_read_at
786 await session.commit()
787 return last_read_at, was_unread
788 return None
789 except Exception:
790 return None
792 async def mark_chat_unread_by_id(
793 self, id: str, user_id: str, db: AsyncSession | None = None
794 ) -> ChatTitleIdResponse | None:
795 try:
796 async with get_async_db_context(db) as session:
797 chat = await session.get(Chat, id)
798 if chat and chat.user_id == user_id:
799 chat.last_read_at = 0
800 await session.commit()
801 return ChatTitleIdResponse(
802 id=chat.id,
803 title=chat.title,
804 updated_at=chat.updated_at,
805 created_at=chat.created_at,
806 last_read_at=chat.last_read_at,
807 )
808 return None
809 except Exception:
810 return None
812 async def mark_chats_read_by_folder_ids(
813 self, user_id: str, folder_ids: list[str], db: AsyncSession | None = None
814 ) -> int:
815 if not folder_ids: 815 ↛ 816line 815 didn't jump to line 816 because the condition on line 815 was never true
816 return 0
818 async with get_async_db_context(db) as session:
819 result = await session.execute(
820 update(Chat)
821 .where(
822 Chat.user_id == user_id,
823 Chat.folder_id.in_(folder_ids),
824 Chat.archived == False,
825 Chat.meta['internal'].as_boolean().is_not(True),
826 )
827 .values(last_read_at=Chat.updated_at)
828 )
829 await session.commit()
830 return result.rowcount or 0
832 async def mark_chats_read_by_user_id(self, user_id: str, db: AsyncSession | None = None) -> int:
833 async with get_async_db_context(db) as session:
834 result = await session.execute(
835 update(Chat)
836 .where(
837 Chat.user_id == user_id,
838 Chat.archived == False,
839 Chat.meta['internal'].as_boolean().is_not(True),
840 )
841 .values(last_read_at=Chat.updated_at)
842 )
843 await session.commit()
844 return result.rowcount or 0
846 async def update_chat_title_by_id(self, id: str, title: str) -> ChatModel | None:
847 try:
848 async with get_async_db_context() as session:
849 chat_item = await session.get(
850 Chat,
851 id,
852 populate_existing=True,
853 with_for_update=session.bind.dialect.name == 'postgresql',
854 )
855 if chat_item is None:
856 return None
857 clean_title = self._clean_null_bytes(title)
858 chat_item.title = clean_title
859 chat_item.chat = {**(chat_item.chat or {}), 'title': clean_title}
860 await session.commit()
861 return ChatModel.model_validate(chat_item)
862 except Exception:
863 return None
865 async def update_chat_tags_by_id(self, id: str, tags: list[str], user) -> None:
866 """Replace a chat's tags. Runs after every completion with tag
867 generation enabled, so only the meta column is read and written,
868 never the chat blob."""
869 async with get_async_db_context() as session:
870 row = (await session.execute(select(Chat.meta).filter_by(id=id))).one_or_none()
871 if row is None:
872 return None
874 meta = row[0] or {}
875 old_tags = meta.get('tags', [])
876 new_tags = [t for t in tags if t.replace(' ', '_').lower() != 'none']
877 new_tag_ids = [t.replace(' ', '_').lower() for t in new_tags]
879 # Single meta update
880 await session.execute(update(Chat).filter_by(id=id).values(meta={**meta, 'tags': new_tag_ids}))
881 await session.commit()
883 # Batch-create any missing tag rows
884 await Tags.ensure_tags_exist(new_tags, user.id, db=session)
886 # Clean up orphaned old tags
887 removed = set(old_tags) - set(new_tag_ids)
888 if removed:
889 await self.delete_orphan_tags_for_user(list(removed), user.id, db=session)
891 async def get_chat_title_by_id(self, id: str) -> str | None:
892 async with get_async_db_context() as session:
893 result = await session.execute(select(Chat.title).filter_by(id=id))
894 row = result.first()
895 if row is None:
896 return None
897 return row[0] or 'New Chat'
899 @staticmethod
900 def get_unresolved_parent_ids(messages_map: dict) -> set[str]:
901 """Return parent IDs referenced by messages but absent from the map.
903 An empty set means the message graph is fully connected.
904 """
905 return {
906 msg['parentId']
907 for msg in messages_map.values()
908 if msg.get('parentId') and msg['parentId'] not in messages_map
909 }
911 @staticmethod
912 def merge_history(existing_history: dict | None, incoming_history: dict | None) -> dict:
913 existing = (existing_history or {}).get('messages') or {}
914 incoming = (incoming_history or {}).get('messages') or {}
915 merged = {
916 message_id: {**message, 'childrenIds': []}
917 for message_id, message in {**existing, **incoming}.items()
918 if isinstance(message, dict)
919 }
921 for message_id, message in merged.items():
922 parent_id = message.get('parentId')
923 if parent_id in merged:
924 merged[parent_id]['childrenIds'].append(message_id)
926 current_id = (incoming_history or {}).get('currentId')
927 if current_id not in merged:
928 current_id = (existing_history or {}).get('currentId')
929 if current_id not in merged:
930 current_id = None
932 return {**(existing_history or {}), **(incoming_history or {}), 'messages': merged, 'currentId': current_id}
934 @staticmethod
935 def delete_message_from_history(history: dict, message_id: str) -> set[str]:
936 messages = history.get('messages') or {}
937 message = messages.get(message_id)
938 if not isinstance(message, dict):
939 return set()
941 parent_id = message.get('parentId')
942 child_ids = [child_id for child_id in (message.get('childrenIds') or []) if child_id in messages]
943 grandchild_ids = [
944 grandchild_id
945 for child_id in child_ids
946 for grandchild_id in (messages.get(child_id, {}).get('childrenIds') or [])
947 if grandchild_id in messages
948 ]
950 if parent_id in messages: 950 ↛ 951line 950 didn't jump to line 951 because the condition on line 950 was never true
951 messages[parent_id]['childrenIds'] = [
952 child_id for child_id in (messages[parent_id].get('childrenIds') or []) if child_id != message_id
953 ] + grandchild_ids
955 for grandchild_id in grandchild_ids: 955 ↛ 956line 955 didn't jump to line 956 because the loop on line 955 never started
956 messages[grandchild_id]['parentId'] = parent_id
958 deleted_ids = {message_id, *child_ids}
959 for deleted_id in deleted_ids:
960 messages.pop(deleted_id, None)
962 current_id = parent_id
963 child_ids = (
964 [child_id for child_id, child in messages.items() if child.get('parentId') is None]
965 if current_id is None
966 else messages.get(current_id, {}).get('childrenIds', [])
967 )
968 visited_ids = set()
969 while child_ids and child_ids[-1] not in visited_ids:
970 current_id = child_ids[-1]
971 visited_ids.add(current_id)
972 child_ids = messages.get(current_id, {}).get('childrenIds', [])
973 history['currentId'] = current_id if current_id in messages else None
974 return deleted_ids
976 @staticmethod
977 def upsert_message_to_history(history: dict, message_id: str, message: dict) -> dict:
978 messages = history.setdefault('messages', {})
980 if message_id in messages:
981 messages[message_id] = {
982 **messages[message_id],
983 **message,
984 }
985 else:
986 message_parent_id = message.get('parentId')
987 parent_id = message_parent_id
988 if parent_id is None: 988 ↛ 994line 988 didn't jump to line 994 because the condition on line 988 was always true
989 for existing_id, existing_message in messages.items():
990 if message_id in existing_message.get('childrenIds', []): 990 ↛ 991line 990 didn't jump to line 991 because the condition on line 990 was never true
991 parent_id = existing_id
992 break
994 parent = messages.get(parent_id) if parent_id else None
995 output = message.get('output') or []
996 output_role = next(
997 (item.get('role') for item in output if isinstance(item, dict) and item.get('role')),
998 None,
999 )
1000 role = message.get('role') or output_role
1001 if not role: 1001 ↛ 1010line 1001 didn't jump to line 1010 because the condition on line 1001 was always true
1002 parent_role = parent.get('role') if parent else None
1003 if parent_role == 'user': 1003 ↛ 1004line 1003 didn't jump to line 1004 because the condition on line 1003 was never true
1004 role = 'assistant'
1005 elif parent_role == 'assistant': 1005 ↛ 1006line 1005 didn't jump to line 1006 because the condition on line 1005 was never true
1006 role = 'user'
1007 else:
1008 role = 'assistant'
1010 messages[message_id] = {
1011 **message,
1012 'id': message.get('id') or message_id,
1013 'parentId': message_parent_id if message_parent_id is not None else parent_id,
1014 'childrenIds': (message.get('childrenIds') if isinstance(message.get('childrenIds'), list) else []),
1015 'role': role,
1016 'timestamp': message.get('timestamp') or int(time.time()),
1017 }
1018 history['currentId'] = message_id
1020 ChatTable._add_child_id_to_parent(messages, messages[message_id].get('parentId'), message_id)
1021 return messages[message_id]
1023 async def backfill_messages_by_chat_id(self, chat_id: str, user_id: str, messages: dict[str, dict]) -> None:
1024 """Write messages to the ``chat_message`` table so future lookups
1025 use the fast path. Errors are logged but never raised.
1026 """
1027 writable = {
1028 message_id: message
1029 for message_id, message in messages.items()
1030 if isinstance(message, dict) and message.get('role')
1031 }
1032 try:
1033 await ChatMessages.upsert_messages(chat_id, user_id, writable)
1034 except Exception as e:
1035 log.warning('Backfill failed for chat %s: %s', chat_id, e)
1037 async def reconcile_messages_by_chat_id(self, chat_id: str, user_id: str, messages: dict[str, dict]) -> None:
1038 """Sync ``chat_message`` rows with the committed JSON blob.
1040 Upserts current messages via ``backfill_messages_by_chat_id``.
1041 Best-effort: errors are logged but never raised.
1042 """
1043 try:
1044 await self.backfill_messages_by_chat_id(chat_id, user_id, messages)
1045 except Exception as e:
1046 log.warning('Failed to reconcile chat_message rows for chat %s: %s', chat_id, e)
1048 async def get_messages_map_by_chat_id(self, id: str) -> dict | None:
1049 """Message map for walking history (see ``get_message_list``).
1051 Prefer ``chat_message`` rows to avoid loading the large embedded
1052 history; fall back to the legacy JSON when no rows exist.
1053 When rows exist but the parent-link graph has gaps (e.g. migration
1054 failures), missing messages are merged from the legacy history
1055 and backfilled so future requests self-heal.
1056 """
1057 # Fast path: build from normalized chat_message rows.
1058 messages_map = await ChatMessages.get_messages_map_by_chat_id(id)
1060 if messages_map is not None:
1061 unresolved_ids = self.get_unresolved_parent_ids(messages_map)
1062 if not unresolved_ids: 1062 ↛ 1066line 1062 didn't jump to line 1066 because the condition on line 1062 was always true
1063 return messages_map
1065 # Graph has gaps — enrich from the legacy embedded history.
1066 log.info(
1067 'Chat %s: %d unresolved parent reference(s) in chat_message — enriching from legacy history',
1068 id,
1069 len(unresolved_ids),
1070 )
1071 chat = await self.get_chat_by_id(id)
1072 if chat:
1073 history_messages = chat.chat.get('history', {}).get('messages', {}) or {}
1074 missing_messages = {
1075 message_id: history_messages[message_id]
1076 for message_id in unresolved_ids
1077 if message_id in history_messages
1078 }
1080 if missing_messages:
1081 messages_map.update(missing_messages)
1083 # Backfill so future requests use the fast path.
1084 await self.backfill_messages_by_chat_id(id, chat.user_id, missing_messages)
1086 return messages_map
1088 # No rows — fall back to the legacy embedded history.
1089 chat = await self.get_chat_by_id(id)
1090 if chat is None: 1090 ↛ 1091line 1090 didn't jump to line 1091 because the condition on line 1090 was never true
1091 return None
1093 history_messages = chat.chat.get('history', {}).get('messages', {}) or {}
1095 # Backfill so future requests use the fast path.
1096 if history_messages: 1096 ↛ 1097line 1096 didn't jump to line 1097 because the condition on line 1096 was never true
1097 await self.backfill_messages_by_chat_id(id, chat.user_id, history_messages)
1099 return history_messages
1101 async def get_message_by_id_and_message_id(self, id: str, message_id: str) -> dict | None:
1102 messages_map = await ChatMessages.get_messages_map_by_chat_id(id)
1103 if messages_map and message_id in messages_map:
1104 return messages_map[message_id]
1106 # Messages the frontend saved straight into the chat blob have no chat_message row yet.
1107 async with get_async_db_context() as session:
1108 result = await session.execute(select(Chat.chat[('history', 'messages')]).filter_by(id=id))
1109 row = result.one_or_none()
1111 if row is None: 1111 ↛ 1112line 1111 didn't jump to line 1112 because the condition on line 1111 was never true
1112 return None
1114 messages = row[0] or {}
1115 return self._clean_null_bytes(messages.get(message_id, {}))
1117 async def get_message_metadata(
1118 self,
1119 chat_id: str,
1120 message_id: str,
1121 metadata_key: Literal['files', 'sources', 'embeds'],
1122 ) -> Any | None:
1123 """Read one message metadata field without rebuilding the whole history."""
1124 async with get_async_db_context() as db:
1125 # Read the column directly; some stored rows cannot be validated as full ChatMessageModel objects.
1126 result = await db.execute(
1127 select(getattr(ChatMessage, metadata_key)).where(ChatMessage.id == f'{chat_id}-{message_id}')
1128 )
1129 metadata_row = result.first()
1131 if metadata_row is not None:
1132 return metadata_row[0]
1134 chat = await self.get_chat_by_id(chat_id)
1135 if chat is None:
1136 return None
1138 message = chat.chat.get('history', {}).get('messages', {}).get(message_id, {})
1139 return message.get(metadata_key)
1141 async def upsert_message_to_chat_by_id_and_message_id(
1142 self, id: str, message_id: str, message: dict, *, touch: bool = True
1143 ) -> ChatModel | None:
1144 if not message.get('content'):
1145 output_text = get_output_text(message.get('output'))
1146 if output_text: 1146 ↛ 1147line 1146 didn't jump to line 1147 because the condition on line 1146 was never true
1147 message['content'] = output_text
1149 message = self._clean_null_bytes(message)
1150 message_id = self._clean_null_bytes(message_id)
1152 try:
1153 async with get_async_db_context() as session:
1154 chat_item = await session.get(
1155 Chat,
1156 id,
1157 populate_existing=True,
1158 with_for_update=session.bind.dialect.name == 'postgresql',
1159 )
1160 if chat_item is None:
1161 return None
1163 chat = chat_item.chat or {}
1164 self._repair_chat_current_id(chat)
1166 history = chat.get('history', {})
1167 saved_message = self.upsert_message_to_history(history, message_id, message)
1168 chat['history'] = history
1169 chat_item.chat = chat # chat is a fresh dict when the column was empty
1170 chat_item.title = self._clean_null_bytes(chat.get('title', 'New Chat'))
1171 chat_item.current_message_id = self.get_current_message_id(chat)
1172 flag_modified(chat_item, 'chat')
1174 if touch:
1175 chat_item.updated_at = int(time.time())
1177 await session.commit()
1178 updated_chat = ChatModel.model_validate(chat_item)
1179 user_id = chat_item.user_id
1181 # Dual-write to chat_message table
1182 try:
1183 await ChatMessages.upsert_message(
1184 message_id=message_id,
1185 chat_id=id,
1186 user_id=user_id,
1187 data=saved_message,
1188 )
1189 except Exception as e:
1190 log.warning(f'Failed to write to chat_message table: {e}')
1192 return updated_chat
1193 except Exception:
1194 return None
1196 async def delete_message_from_chat_by_id_and_message_id(self, id: str, message_id: str) -> ChatModel | None:
1197 try:
1198 async with get_async_db_context() as session:
1199 chat_item = await session.get(
1200 Chat,
1201 id,
1202 populate_existing=True,
1203 with_for_update=session.bind.dialect.name == 'postgresql',
1204 )
1205 if chat_item is None:
1206 return None
1208 chat = chat_item.chat or {}
1209 self._repair_chat_current_id(chat)
1211 history = chat.get('history', {})
1212 deleted_ids = self.delete_message_from_history(history, message_id)
1213 if not deleted_ids:
1214 chat_item.chat = chat
1215 chat_item.title = self._clean_null_bytes(chat.get('title', 'New Chat'))
1216 chat_item.current_message_id = self.get_current_message_id(chat)
1217 flag_modified(chat_item, 'chat')
1218 await session.commit()
1219 return ChatModel.model_validate(chat_item)
1221 messages = history.get('messages') or {}
1222 chat['history'] = history
1223 chat_item.chat = chat
1224 chat_item.title = self._clean_null_bytes(chat.get('title', 'New Chat'))
1225 chat_item.current_message_id = self.get_current_message_id(chat)
1226 flag_modified(chat_item, 'chat')
1227 chat_item.updated_at = int(time.time())
1228 await session.commit()
1229 updated_chat = ChatModel.model_validate(chat_item)
1230 user_id = chat_item.user_id
1232 await self.backfill_messages_by_chat_id(id, user_id, messages)
1233 await ChatMessages.delete_message_ids_by_chat_id(id, deleted_ids)
1235 return updated_chat
1236 except Exception:
1237 return None
1239 async def add_message_status_to_chat_by_id_and_message_id(
1240 self, id: str, message_id: str, status: dict
1241 ) -> ChatModel | None:
1242 try:
1243 status = self._clean_null_bytes(status)
1244 async with get_async_db_context() as session:
1245 chat_item = await session.get(
1246 Chat,
1247 id,
1248 populate_existing=True,
1249 with_for_update=session.bind.dialect.name == 'postgresql',
1250 )
1251 if chat_item is None:
1252 return None
1254 chat = chat_item.chat or {}
1255 self._repair_chat_current_id(chat)
1256 history = chat.get('history', {})
1258 if message_id in history.get('messages', {}):
1259 status_history = history['messages'][message_id].get('statusHistory', [])
1260 status_history.append(status)
1261 history['messages'][message_id]['statusHistory'] = status_history
1263 chat['history'] = history
1264 chat_item.chat = chat
1265 chat_item.title = self._clean_null_bytes(chat.get('title', 'New Chat'))
1266 chat_item.current_message_id = self.get_current_message_id(chat)
1267 flag_modified(chat_item, 'chat')
1268 await session.commit()
1270 return ChatModel.model_validate(chat_item)
1271 except Exception:
1272 return None
1274 async def add_message_files_by_id_and_message_id(
1275 self, id: str, message_id: str, files: list[dict]
1276 ) -> list[dict] | None:
1277 async with get_async_db_context() as session:
1278 chat_item = await session.get(
1279 Chat,
1280 id,
1281 populate_existing=True,
1282 with_for_update=session.bind.dialect.name == 'postgresql',
1283 )
1284 if chat_item is None:
1285 return None
1287 chat = chat_item.chat or {}
1288 history = chat.get('history', {})
1290 message_files = []
1292 if message_id in history.get('messages', {}):
1293 message_files = history['messages'][message_id].get('files', [])
1294 message_files = message_files + files
1295 history['messages'][message_id]['files'] = message_files
1297 # Written here rather than through update_chat_by_id: with session sharing off that opens a second
1298 # connection, which then blocks on the lock this one holds.
1299 chat['history'] = history
1300 chat_item.chat = self._clean_null_bytes(chat)
1301 # History was mutated in place, so the new blob compares equal to the loaded one.
1302 flag_modified(chat_item, 'chat')
1303 chat_item.updated_at = int(time.time())
1304 await session.commit()
1305 return message_files
1307 async def insert_shared_chat_by_chat_id(self, chat_id: str, db: AsyncSession | None = None) -> ChatModel | None:
1308 """Create a shared snapshot for a chat. Returns the original chat with share_id set."""
1309 from open_webui.models.shared_chats import SharedChats
1311 async with get_async_db_context(db) as session:
1312 chat = await session.get(Chat, chat_id)
1313 if not chat:
1314 return None
1316 # If already shared, just update the existing snapshot
1317 if chat.share_id:
1318 return await self.update_shared_chat_by_chat_id(chat_id, db=session)
1320 shared = await SharedChats.create(chat_id, chat.user_id, db=session)
1321 if not shared:
1322 return None
1324 # Set share_id on the original chat
1325 chat.share_id = shared.id
1326 await session.commit()
1327 return ChatModel.model_validate(chat) # return the updated original
1329 # refresh helper
1330 async def update_shared_chat_by_chat_id(
1331 self,
1332 chat_id: str,
1333 db: AsyncSession | None = None,
1334 ) -> ChatModel | None:
1335 """Refresh the shared snapshot with current chat content."""
1336 from open_webui.models.shared_chats import SharedChats
1338 async with get_async_db_context(db) as session:
1339 record = await session.get(Chat, chat_id)
1340 if not record or not record.share_id:
1341 return await self.insert_shared_chat_by_chat_id(chat_id, db=session)
1342 await SharedChats.update(record.share_id, db=session)
1343 return ChatModel.model_validate(record)
1344 # unreachable — context manager above always returns
1345 return
1347 async def delete_shared_chat_by_chat_id(self, chat_id: str, db: AsyncSession | None = None) -> bool:
1348 """Delete shared snapshot for a chat."""
1349 from open_webui.models.shared_chats import SharedChats
1351 try:
1352 return await SharedChats.delete_by_chat_id(chat_id, db=db)
1353 except Exception:
1354 return False
1356 async def unarchive_all_chats_by_user_id(self, user_id: str, db: AsyncSession | None = None) -> bool:
1357 try:
1358 async with get_async_db_context(db) as session:
1359 await session.execute(update(Chat).filter_by(user_id=user_id).values(archived=False))
1360 await session.commit()
1361 return True
1362 except Exception:
1363 return False
1365 async def update_chat_share_id_by_id(
1366 self, id: str, share_id: str | None, db: AsyncSession | None = None
1367 ) -> ChatModel | None:
1368 try:
1369 async with get_async_db_context(db) as session:
1370 chat = await session.get(Chat, id)
1371 chat.share_id = share_id
1372 await session.commit()
1373 return ChatModel.model_validate(chat)
1374 except Exception:
1375 return None
1377 async def toggle_chat_pinned_by_id(self, id: str, db: AsyncSession | None = None) -> ChatModel | None:
1378 try:
1379 async with get_async_db_context(db) as session:
1380 chat = await session.get(Chat, id)
1381 chat.pinned = not chat.pinned
1382 chat.updated_at = int(time.time())
1383 chat.last_read_at = int(time.time())
1384 await session.commit()
1385 return ChatModel.model_validate(chat)
1386 except Exception:
1387 return None
1389 async def toggle_chat_archive_by_id(self, id: str, db: AsyncSession | None = None) -> ChatModel | None:
1390 try:
1391 async with get_async_db_context(db) as session:
1392 chat = await session.get(Chat, id)
1393 chat.archived = not chat.archived
1394 chat.folder_id = None
1395 chat.updated_at = int(time.time())
1396 chat.last_read_at = int(time.time())
1397 await session.commit()
1398 return ChatModel.model_validate(chat)
1399 except Exception:
1400 return None
1402 async def archive_all_chats_by_user_id(self, user_id: str, db: AsyncSession | None = None) -> bool:
1403 try:
1404 async with get_async_db_context(db) as session:
1405 await session.execute(update(Chat).filter_by(user_id=user_id).values(archived=True))
1406 await session.commit()
1407 return True
1408 except Exception:
1409 return False
1411 async def get_archived_chat_list_by_user_id(
1412 self,
1413 user_id: str,
1414 filter: dict | None = None,
1415 skip: int = 0,
1416 limit: int = 50,
1417 db: AsyncSession | None = None,
1418 ) -> list[ChatTitleIdResponse]:
1419 async with get_async_db_context(db) as session:
1420 stmt = select(Chat.id, Chat.title, Chat.updated_at, Chat.created_at).filter_by(
1421 user_id=user_id, archived=True
1422 )
1423 stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True))
1425 if filter:
1426 query_key = filter.get('query')
1427 if query_key:
1428 stmt = stmt.filter(Chat.title.ilike(f'%{query_key}%'))
1430 order_by = filter.get('order_by')
1431 direction = filter.get('direction')
1433 if order_by and direction:
1434 if not getattr(Chat, order_by, None): 1434 ↛ 1437line 1434 didn't jump to line 1437 because the condition on line 1434 was always true
1435 raise ValueError('Invalid order_by field')
1437 if direction.lower() == 'asc':
1438 stmt = stmt.order_by(getattr(Chat, order_by).asc(), Chat.id)
1439 elif direction.lower() == 'desc':
1440 stmt = stmt.order_by(getattr(Chat, order_by).desc(), Chat.id)
1441 else:
1442 raise ValueError('Invalid direction for ordering')
1443 else:
1444 stmt = stmt.order_by(Chat.updated_at.desc(), Chat.id)
1446 if skip:
1447 stmt = stmt.offset(skip)
1448 if limit: 1448 ↛ 1451line 1448 didn't jump to line 1451 because the condition on line 1448 was always true
1449 stmt = stmt.limit(limit)
1451 result = await session.execute(stmt)
1452 all_chats = result.all()
1453 return [
1454 ChatTitleIdResponse.model_validate(
1455 {
1456 'id': chat[0],
1457 'title': chat[1],
1458 'updated_at': chat[2],
1459 'created_at': chat[3],
1460 }
1461 )
1462 for chat in all_chats
1463 ]
1465 async def count_archived_chats_by_user_id(
1466 self,
1467 user_id: str,
1468 db: AsyncSession | None = None,
1469 ) -> int:
1470 async with get_async_db_context(db) as session:
1471 stmt = select(func.count(Chat.id)).filter_by(user_id=user_id, archived=True)
1472 result = await session.execute(stmt.where(Chat.meta['internal'].as_boolean().is_not(True)))
1473 return result.scalar() or 0
1475 async def get_shared_chat_list_by_user_id(
1476 self,
1477 user_id: str,
1478 filter: dict | None = None,
1479 skip: int = 0,
1480 limit: int = 50,
1481 db: AsyncSession | None = None,
1482 ) -> list[SharedChatResponse]:
1483 """Delegate to SharedChats for listing shared chats by user."""
1484 from open_webui.models.shared_chats import SharedChats
1486 return await SharedChats.get_by_user_id(user_id, filter=filter, skip=skip, limit=limit, db=db)
1488 async def get_chat_list_by_user_id(
1489 self,
1490 user_id: str,
1491 include_archived: bool = False,
1492 filter: dict | None = None,
1493 skip: int = 0,
1494 limit: int = 50,
1495 db: AsyncSession | None = None,
1496 ) -> list[ChatTitleIdResponse]:
1497 async with get_async_db_context(db) as session:
1498 stmt = select(Chat.id, Chat.title, Chat.updated_at, Chat.created_at, Chat.last_read_at).filter_by(
1499 user_id=user_id
1500 )
1501 stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True))
1502 if not include_archived:
1503 stmt = stmt.filter_by(archived=False)
1505 if filter:
1506 query_key = filter.get('query')
1507 if query_key:
1508 stmt = stmt.filter(Chat.title.ilike(f'%{query_key}%'))
1510 order_by = filter.get('order_by')
1511 direction = filter.get('direction')
1513 if order_by and direction and getattr(Chat, order_by): 1513 ↛ 1514line 1513 didn't jump to line 1514 because the condition on line 1513 was never true
1514 if direction.lower() == 'asc':
1515 stmt = stmt.order_by(getattr(Chat, order_by).asc(), Chat.id)
1516 elif direction.lower() == 'desc':
1517 stmt = stmt.order_by(getattr(Chat, order_by).desc(), Chat.id)
1518 else:
1519 raise ValueError('Invalid direction for ordering')
1520 else:
1521 stmt = stmt.order_by(Chat.updated_at.desc(), Chat.id)
1523 if skip:
1524 stmt = stmt.offset(skip)
1525 if limit: 1525 ↛ 1528line 1525 didn't jump to line 1528 because the condition on line 1525 was always true
1526 stmt = stmt.limit(limit)
1528 result = await session.execute(stmt)
1529 all_chats = result.all()
1530 return [
1531 ChatTitleIdResponse.model_validate(
1532 {
1533 'id': chat[0],
1534 'title': chat[1],
1535 'updated_at': chat[2],
1536 'created_at': chat[3],
1537 'last_read_at': chat[4],
1538 }
1539 )
1540 for chat in all_chats
1541 ]
1543 async def get_chat_title_id_list_by_user_id(
1544 self,
1545 user_id: str,
1546 include_archived: bool = False,
1547 include_folders: bool = False,
1548 include_pinned: bool = False,
1549 sort_by: str = 'updated_at',
1550 sort_dir: str = 'desc',
1551 skip: int | None = None,
1552 limit: int | None = None,
1553 db: AsyncSession | None = None,
1554 ) -> list[ChatTitleIdResponse]:
1555 async with get_async_db_context(db) as session:
1556 stmt = select(Chat.id, Chat.title, Chat.updated_at, Chat.created_at, Chat.last_read_at).filter_by(
1557 user_id=user_id
1558 )
1559 stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True))
1561 if not include_folders:
1562 stmt = stmt.filter_by(folder_id=None)
1564 if not include_pinned:
1565 stmt = stmt.filter(or_(Chat.pinned == False, Chat.pinned == None))
1567 if not include_archived: 1567 ↛ 1570line 1567 didn't jump to line 1570 because the condition on line 1567 was always true
1568 stmt = stmt.filter_by(archived=False)
1570 stmt = stmt.order_by(*chat_list_order(sort_by, sort_dir))
1572 if skip:
1573 stmt = stmt.offset(skip)
1574 if limit:
1575 stmt = stmt.limit(limit)
1577 result = await session.execute(stmt)
1578 all_chats = result.all()
1580 return [
1581 ChatTitleIdResponse.model_validate(
1582 {
1583 'id': chat[0],
1584 'title': chat[1],
1585 'updated_at': chat[2],
1586 'created_at': chat[3],
1587 'last_read_at': chat[4],
1588 }
1589 )
1590 for chat in all_chats
1591 ]
1593 async def get_chat_list_by_chat_ids(
1594 self,
1595 chat_ids: list[str],
1596 skip: int = 0,
1597 limit: int = 50,
1598 db: AsyncSession | None = None,
1599 ) -> list[ChatModel]:
1600 async with get_async_db_context(db) as session:
1601 stmt = select(Chat).filter(Chat.id.in_(chat_ids)).filter_by(archived=False)
1602 stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True))
1603 result = await session.execute(stmt.order_by(Chat.updated_at.desc()))
1604 all_chats = result.scalars().all()
1605 return [ChatModel.model_validate(chat) for chat in all_chats]
1607 async def get_chat_metas_by_chat_ids(
1608 self,
1609 chat_ids: list[str],
1610 include_archived: bool = False,
1611 db: AsyncSession | None = None,
1612 ) -> list[dict]:
1613 async with get_async_db_context(db) as session:
1614 stmt = select(Chat.meta).filter(Chat.id.in_(chat_ids))
1615 if not include_archived:
1616 stmt = stmt.filter_by(archived=False)
1618 result = await session.execute(stmt)
1619 return [meta for meta in result.scalars().all() if isinstance(meta, dict)]
1621 async def get_chats_by_model_id(
1622 self,
1623 model_id: str,
1624 filter: dict | None = None,
1625 skip: int = 0,
1626 limit: int = 50,
1627 db: AsyncSession | None = None,
1628 ) -> dict:
1629 from open_webui.models.users import User
1631 async with get_async_db_context(db) as session:
1632 chat_ids = (
1633 select(ChatMessage.chat_id).filter(ChatMessage.model_id == model_id).group_by(ChatMessage.chat_id)
1634 )
1636 if filter:
1637 if filter.get('start_date'):
1638 chat_ids = chat_ids.filter(ChatMessage.created_at >= filter.get('start_date'))
1639 if filter.get('end_date'):
1640 chat_ids = chat_ids.filter(ChatMessage.created_at <= filter.get('end_date'))
1642 chat_ids = chat_ids.subquery()
1644 stmt = (
1645 select(Chat.id, Chat.user_id, Chat.title, Chat.updated_at, User.name.label('user_name'))
1646 .join(chat_ids, chat_ids.c.chat_id == Chat.id)
1647 .outerjoin(User, User.id == Chat.user_id)
1648 .where(Chat.meta['internal'].as_boolean().is_not(True))
1649 )
1651 order_by = filter.get('order_by') if filter else None
1652 direction = filter.get('direction') if filter else None
1653 is_asc = direction == 'asc'
1655 if order_by == 'title': 1655 ↛ 1656line 1655 didn't jump to line 1656 because the condition on line 1655 was never true
1656 primary_sort = Chat.title.asc() if is_asc else Chat.title.desc()
1657 elif order_by == 'user_name': 1657 ↛ 1658line 1657 didn't jump to line 1658 because the condition on line 1657 was never true
1658 primary_sort = User.name.asc() if is_asc else User.name.desc()
1659 else:
1660 primary_sort = Chat.updated_at.asc() if is_asc else Chat.updated_at.desc()
1662 stmt = stmt.order_by(primary_sort, Chat.id.asc())
1664 count_result = await session.execute(select(func.count()).select_from(stmt.subquery()))
1665 total = count_result.scalar()
1667 if skip:
1668 stmt = stmt.offset(skip)
1669 if limit:
1670 stmt = stmt.limit(limit)
1672 result = await session.execute(stmt)
1673 return {
1674 'items': [
1675 {
1676 'chat_id': chat.id,
1677 'user_id': chat.user_id,
1678 'user_name': chat.user_name,
1679 'first_message': chat.title,
1680 'updated_at': chat.updated_at,
1681 }
1682 for chat in result.all()
1683 ],
1684 'total': total,
1685 }
1687 # retrieve conversation
1688 async def get_chat_by_id(
1689 self,
1690 id: str,
1691 db: AsyncSession | None = None,
1692 ) -> ChatModel | None:
1693 """Fetch a chat by PK, auto-sanitizing null bytes on read."""
1694 try:
1695 async with get_async_db_context(db) as session:
1696 chat_item = await session.get(Chat, id)
1697 if chat_item is None:
1698 return None
1700 repaired_history = self._repair_chat_current_id(chat_item.chat or {})
1701 if repaired_history:
1702 chat_item.current_message_id = self.get_current_message_id(chat_item.chat)
1703 flag_modified(chat_item, 'chat')
1704 if self._sanitize_chat_row(chat_item) or repaired_history:
1705 await session.commit()
1707 return ChatModel.model_validate(chat_item)
1708 except Exception:
1709 return None
1711 async def get_chat_by_share_id(self, id: str, db: AsyncSession | None = None) -> ChatModel | None:
1712 """Look up a shared chat snapshot by its share token."""
1713 from open_webui.models.shared_chats import SharedChats
1715 try:
1716 shared = await SharedChats.get_by_id(id, db=db)
1717 if shared:
1718 # Return a ChatModel-compatible view of the snapshot
1719 return ChatModel(
1720 id=shared.id,
1721 user_id=shared.user_id,
1722 title=shared.title,
1723 chat=shared.chat,
1724 created_at=shared.created_at,
1725 updated_at=shared.updated_at,
1726 share_id=shared.id,
1727 )
1728 return None
1729 except Exception:
1730 return None
1732 async def get_chat_by_id_and_user_id(
1733 self, id: str, user_id: str, db: AsyncSession | None = None
1734 ) -> ChatModel | None:
1735 try:
1736 async with get_async_db_context(db) as session:
1737 result = await session.execute(select(Chat).filter_by(id=id, user_id=user_id))
1738 chat = result.scalars().first()
1739 if not chat:
1740 return None
1742 repaired_history = self._repair_chat_current_id(chat.chat or {})
1743 if repaired_history:
1744 chat.current_message_id = self.get_current_message_id(chat.chat)
1745 flag_modified(chat, 'chat')
1746 if self._sanitize_chat_row(chat) or repaired_history:
1747 await session.commit()
1749 return ChatModel.model_validate(chat)
1750 except Exception:
1751 return None
1753 async def get_chat_by_id_for_user(
1754 self,
1755 id: str,
1756 user,
1757 db: AsyncSession | None = None,
1758 ) -> ChatModel | None:
1759 chat = await self.get_chat_by_id_and_user_id(id, user.id, db=db)
1760 if chat:
1761 return chat
1763 chat = await self.get_chat_by_id(id, db=db)
1764 if not chat: 1764 ↛ 1767line 1764 didn't jump to line 1767 because the condition on line 1764 was always true
1765 return None
1767 if user.role == 'admin' and (ENABLE_ADMIN_CHAT_ACCESS or is_internal_chat(chat.meta)):
1768 return chat
1770 if await AccessGrants.has_access(
1771 user_id=user.id,
1772 resource_type='shared_chat',
1773 resource_id=id,
1774 permission='read',
1775 db=db,
1776 ):
1777 return chat
1779 if chat.folder_id:
1780 from open_webui.utils.access_control.folders import has_folder_access
1782 folder = await Folders.get_folder_by_id(chat.folder_id, db=db)
1783 if folder and await has_folder_access(user.id, folder, 'read', db):
1784 return chat
1786 return None
1788 async def is_chat_owner(self, id: str, user_id: str, db: AsyncSession | None = None) -> bool:
1789 """
1790 Lightweight ownership check — uses EXISTS subquery instead of loading
1791 the full Chat row (which includes the potentially large JSON blob).
1792 """
1793 try:
1794 async with get_async_db_context(db) as session:
1795 result = await session.execute(select(exists().where(and_(Chat.id == id, Chat.user_id == user_id))))
1796 return result.scalar()
1797 except Exception:
1798 return False
1800 async def get_chat_folder_id(self, id: str, user_id: str, db: AsyncSession | None = None) -> str | None:
1801 """
1802 Fetch only the folder_id column for a chat, without loading the full
1803 JSON blob. Returns None if chat doesn't exist or doesn't belong to user.
1804 """
1805 try:
1806 async with get_async_db_context(db) as session:
1807 result = await session.execute(select(Chat.folder_id).filter_by(id=id, user_id=user_id))
1808 row = result.first()
1809 return row[0] if row else None
1810 except Exception:
1811 return None
1813 async def count_unread_by_folder_ids(
1814 self,
1815 user_id: str,
1816 folder_ids: list[str],
1817 db: AsyncSession | None = None,
1818 ) -> dict[str, int]:
1819 if not folder_ids:
1820 return {}
1822 unfinished_assistant = (
1823 select(ChatMessage.id)
1824 .where(ChatMessage.chat_id == Chat.id)
1825 .where(ChatMessage.role == 'assistant')
1826 .where(ChatMessage.done.is_(False))
1827 .exists()
1828 )
1830 async with get_async_db_context(db) as session:
1831 result = await session.execute(
1832 select(Chat.folder_id, func.count(Chat.id))
1833 .where(
1834 Chat.user_id == user_id,
1835 Chat.folder_id.in_(folder_ids),
1836 Chat.archived == False,
1837 Chat.updated_at > func.coalesce(Chat.last_read_at, 0),
1838 ~unfinished_assistant,
1839 )
1840 .group_by(Chat.folder_id)
1841 )
1842 return {folder_id: count for folder_id, count in result.all() if folder_id}
1844 async def get_chats(self, skip: int = 0, limit: int = 50, db: AsyncSession | None = None) -> list[ChatModel]:
1845 async with get_async_db_context(db) as session:
1846 stmt = select(Chat).where(Chat.meta['internal'].as_boolean().is_not(True))
1847 result = await session.execute(stmt.order_by(Chat.updated_at.desc()))
1848 all_chats = result.scalars().all()
1849 return [ChatModel.model_validate(chat) for chat in all_chats]
1851 async def get_user_usage_chat_stats(self, user_id: str, db: AsyncSession | None = None) -> dict:
1852 async with get_async_db_context(db) as session:
1853 chat_filter = (Chat.user_id == user_id, Chat.meta['internal'].as_boolean().is_not(True))
1854 result = await session.execute(select(func.count(Chat.id).label('total_chats')).where(*chat_filter))
1855 total_chats = int(result.scalar() or 0)
1857 messages_stmt = (
1858 select(ChatMessage.chat_id, ChatMessage.created_at)
1859 .join(Chat, Chat.id == ChatMessage.chat_id)
1860 .where(*chat_filter, ChatMessage.created_at.isnot(None))
1861 .order_by(ChatMessage.chat_id, ChatMessage.created_at.asc())
1862 )
1863 messages_result = await session.execute(messages_stmt)
1864 last_message_at_by_chat: dict[str, int] = {}
1865 active_seconds_by_chat: dict[str, int] = {}
1867 for chat_id, created_at in messages_result.all():
1868 timestamp = int(created_at / 1000) if created_at > 10_000_000_000 else int(created_at)
1869 last_message_at = last_message_at_by_chat.get(chat_id)
1870 if last_message_at is not None:
1871 delta = timestamp - last_message_at
1872 if 0 < delta <= ACTIVE_CHAT_GAP_SECONDS:
1873 active_seconds_by_chat[chat_id] = active_seconds_by_chat.get(chat_id, 0) + delta
1874 last_message_at_by_chat[chat_id] = timestamp
1876 return {
1877 'total_chats': total_chats,
1878 'longest_chat_seconds': max(active_seconds_by_chat.values(), default=0),
1879 }
1881 # list user conversations
1882 async def get_chats_by_user_id(
1883 self,
1884 user_id: str,
1885 filter: dict | None = None,
1886 skip: int | None = None,
1887 limit: int | None = None,
1888 db: AsyncSession | None = None,
1889 ) -> ChatListResponse:
1890 async with get_async_db_context(db) as session:
1891 stmt = select(Chat).filter_by(user_id=user_id)
1892 stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True))
1894 if filter:
1895 if filter.get('updated_at'):
1896 stmt = stmt.filter(Chat.updated_at > filter.get('updated_at'))
1898 order_by = filter.get('order_by')
1899 direction = filter.get('direction')
1901 if order_by and direction: 1901 ↛ 1908line 1901 didn't jump to line 1908 because the condition on line 1901 was always true
1902 if hasattr(Chat, order_by): 1902 ↛ 1913line 1902 didn't jump to line 1913 because the condition on line 1902 was always true
1903 if direction.lower() == 'asc': 1903 ↛ 1905line 1903 didn't jump to line 1905 because the condition on line 1903 was always true
1904 stmt = stmt.order_by(getattr(Chat, order_by).asc(), Chat.id)
1905 elif direction.lower() == 'desc':
1906 stmt = stmt.order_by(getattr(Chat, order_by).desc(), Chat.id)
1907 else:
1908 stmt = stmt.order_by(Chat.updated_at.desc(), Chat.id)
1910 else:
1911 stmt = stmt.order_by(Chat.updated_at.desc(), Chat.id)
1913 count_result = await session.execute(select(func.count()).select_from(stmt.subquery()))
1914 total = count_result.scalar()
1916 if skip is not None: 1916 ↛ 1918line 1916 didn't jump to line 1918 because the condition on line 1916 was always true
1917 stmt = stmt.offset(skip)
1918 if limit is not None: 1918 ↛ 1921line 1918 didn't jump to line 1921 because the condition on line 1918 was always true
1919 stmt = stmt.limit(limit)
1921 result = await session.execute(stmt)
1922 all_chats = result.scalars().all()
1924 return ChatListResponse(
1925 **{
1926 'items': [ChatModel.model_validate(chat) for chat in all_chats],
1927 'total': total,
1928 }
1929 )
1931 # list pinned chats
1932 async def get_pinned_chats_by_user_id(
1933 self, user_id: str, db: AsyncSession | None = None
1934 ) -> list[ChatTitleIdResponse]:
1935 async with get_async_db_context(db) as session:
1936 stmt = select(Chat.id, Chat.title, Chat.updated_at, Chat.created_at, Chat.last_read_at).filter_by(
1937 user_id=user_id, pinned=True, archived=False
1938 )
1939 stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True))
1940 result = await session.execute(stmt.order_by(Chat.updated_at.desc()))
1941 all_chats = result.all()
1942 return [
1943 ChatTitleIdResponse.model_validate(
1944 {
1945 'id': chat[0],
1946 'title': chat[1],
1947 'updated_at': chat[2],
1948 'created_at': chat[3],
1949 'last_read_at': chat[4],
1950 }
1951 )
1952 for chat in all_chats
1953 ]
1955 async def get_archived_chats_by_user_id(self, user_id: str, db: AsyncSession | None = None) -> list[ChatModel]:
1956 async with get_async_db_context(db) as session:
1957 stmt = select(Chat).filter_by(user_id=user_id, archived=True)
1958 stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True))
1959 result = await session.execute(stmt.order_by(Chat.updated_at.desc()))
1960 return [ChatModel.model_validate(chat) for chat in result.scalars().all()]
1962 # search user conversations
1963 async def get_chats_by_user_id_and_search_text( # noqa: C901
1964 self,
1965 user_id: str,
1966 search_text: str,
1967 include_archived: bool = False,
1968 skip: int = 0,
1969 limit: int = 60,
1970 db: AsyncSession | None = None,
1971 ) -> list[ChatModel]:
1972 """
1973 Filters chats based on a search query using Python, allowing pagination using skip and limit.
1974 """
1975 search_text = sanitize_text_for_db(search_text).lower().strip()
1977 if not search_text:
1978 return await self.get_chat_list_by_user_id(
1979 user_id, include_archived, filter={}, skip=skip, limit=limit, db=db
1980 )
1982 search_text_words = search_text.split()
1984 # search_text might contain 'tag:tag_name' format so we need to extract the tag_name
1985 tag_ids = [
1986 word.replace('tag:', '').replace(' ', '_').lower() for word in search_text_words if word.startswith('tag:')
1987 ]
1989 # Extract folder names
1990 folder_names = [word.replace('folder:', '') for word in search_text_words if word.startswith('folder:')]
1991 folders = await Folders.search_folders_by_names(user_id, folder_names)
1992 folder_ids = [folder.id for folder in folders]
1994 is_pinned = None
1995 if 'pinned:true' in search_text_words: 1995 ↛ 1996line 1995 didn't jump to line 1996 because the condition on line 1995 was never true
1996 is_pinned = True
1997 elif 'pinned:false' in search_text_words: 1997 ↛ 1998line 1997 didn't jump to line 1998 because the condition on line 1997 was never true
1998 is_pinned = False
2000 is_archived = None
2001 if 'archived:true' in search_text_words: 2001 ↛ 2002line 2001 didn't jump to line 2002 because the condition on line 2001 was never true
2002 is_archived = True
2003 elif 'archived:false' in search_text_words: 2003 ↛ 2004line 2003 didn't jump to line 2004 because the condition on line 2003 was never true
2004 is_archived = False
2006 is_shared = None
2007 if 'shared:true' in search_text_words: 2007 ↛ 2008line 2007 didn't jump to line 2008 because the condition on line 2007 was never true
2008 is_shared = True
2009 elif 'shared:false' in search_text_words: 2009 ↛ 2010line 2009 didn't jump to line 2010 because the condition on line 2009 was never true
2010 is_shared = False
2012 search_text_words = [word for word in search_text_words if not word.startswith(CHAT_SEARCH_FILTER_PREFIXES)]
2014 phrase_query = ' '.join(search_text_words).strip()
2015 search_terms = chat_search_terms(phrase_query)
2017 async with get_async_db_context(db) as session:
2018 stmt = select(Chat).filter(Chat.user_id == user_id)
2019 stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True))
2021 if is_archived is not None: 2021 ↛ 2022line 2021 didn't jump to line 2022 because the condition on line 2021 was never true
2022 stmt = stmt.filter(Chat.archived == is_archived)
2023 elif not include_archived: 2023 ↛ 2026line 2023 didn't jump to line 2026 because the condition on line 2023 was always true
2024 stmt = stmt.filter(Chat.archived == False)
2026 if is_pinned is not None: 2026 ↛ 2027line 2026 didn't jump to line 2027 because the condition on line 2026 was never true
2027 stmt = stmt.filter(Chat.pinned == is_pinned)
2029 if is_shared is not None: 2029 ↛ 2030line 2029 didn't jump to line 2030 because the condition on line 2029 was never true
2030 if is_shared:
2031 stmt = stmt.filter(Chat.share_id.isnot(None))
2032 else:
2033 stmt = stmt.filter(Chat.share_id.is_(None))
2035 if folder_names: 2035 ↛ 2036line 2035 didn't jump to line 2036 because the condition on line 2035 was never true
2036 stmt = stmt.filter(Chat.folder_id.in_(folder_ids))
2038 # Check if the database dialect is either 'sqlite' or 'postgresql'
2039 bind = await session.connection()
2040 dialect_name = bind.dialect.name
2042 search_params = {}
2043 exact_match_clause = None
2044 if phrase_query: 2044 ↛ 2074line 2044 didn't jump to line 2074 because the condition on line 2044 was always true
2045 exact_match_clause = or_(
2046 Chat.title.ilike(bindparam('phrase_title_key')),
2047 text(chat_search_message_content_match_sql(dialect_name, 'phrase_content_key')),
2048 )
2049 search_params.update(
2050 {
2051 'phrase_title_key': f'%{phrase_query}%',
2052 'phrase_content_key': phrase_query,
2053 }
2054 )
2056 term_clauses = []
2057 for term_idx, term in enumerate(search_terms):
2058 title_key = f'term_title_key_{term_idx}'
2059 content_key = f'term_content_key_{term_idx}'
2060 term_clauses.append(
2061 or_(
2062 Chat.title.ilike(bindparam(title_key)),
2063 text(chat_search_message_content_match_sql(dialect_name, content_key)),
2064 )
2065 )
2066 search_params[title_key] = f'%{term}%'
2067 search_params[content_key] = term
2069 if term_clauses:
2070 stmt = stmt.filter(or_(exact_match_clause, and_(*term_clauses)))
2071 else:
2072 stmt = stmt.filter(exact_match_clause)
2074 if dialect_name == 'sqlite': 2074 ↛ 2101line 2074 didn't jump to line 2101 because the condition on line 2074 was always true
2075 # Check if there are any tags to filter
2076 if 'none' in tag_ids: 2076 ↛ 2077line 2076 didn't jump to line 2077 because the condition on line 2076 was never true
2077 stmt = stmt.filter(
2078 text("""
2079 NOT EXISTS (
2080 SELECT 1
2081 FROM json_each(Chat.meta, '$.tags') AS tag
2082 )
2083 """)
2084 )
2085 elif tag_ids: 2085 ↛ 2086line 2085 didn't jump to line 2086 because the condition on line 2085 was never true
2086 stmt = stmt.filter(
2087 and_(
2088 *[
2089 text(f"""
2090 EXISTS (
2091 SELECT 1
2092 FROM json_each(Chat.meta, '$.tags') AS tag
2093 WHERE tag.value = :tag_id_{tag_idx}
2094 )
2095 """).params(**{f'tag_id_{tag_idx}': tag_id})
2096 for tag_idx, tag_id in enumerate(tag_ids)
2097 ]
2098 )
2099 )
2101 elif dialect_name == 'postgresql':
2102 # Safety filter: JSON field must not contain \u0000
2103 stmt = stmt.filter(text("Chat.chat::text NOT LIKE '%\\\\u0000%'"))
2105 # Safety filter: title must not contain actual null bytes
2106 stmt = stmt.filter(text("Chat.title::text NOT LIKE '%\\x00%'"))
2108 if 'none' in tag_ids:
2109 stmt = stmt.filter(
2110 text("""
2111 NOT EXISTS (
2112 SELECT 1
2113 FROM json_array_elements_text(Chat.meta->'tags') AS tag
2114 )
2115 """)
2116 )
2117 elif tag_ids:
2118 stmt = stmt.filter(
2119 and_(
2120 *[
2121 text(f"""
2122 EXISTS (
2123 SELECT 1
2124 FROM json_array_elements_text(Chat.meta->'tags') AS tag
2125 WHERE tag = :tag_id_{tag_idx}
2126 )
2127 """).params(**{f'tag_id_{tag_idx}': tag_id})
2128 for tag_idx, tag_id in enumerate(tag_ids)
2129 ]
2130 )
2131 )
2132 else:
2133 raise NotImplementedError(f'Unsupported dialect: {dialect_name}')
2135 if exact_match_clause is not None: 2135 ↛ 2138line 2135 didn't jump to line 2138 because the condition on line 2135 was always true
2136 stmt = stmt.order_by(case((exact_match_clause, 0), else_=1), Chat.updated_at.desc(), Chat.id)
2137 else:
2138 stmt = stmt.order_by(Chat.updated_at.desc(), Chat.id)
2140 if search_params: 2140 ↛ 2144line 2140 didn't jump to line 2144 because the condition on line 2140 was always true
2141 stmt = stmt.params(**search_params)
2143 # Perform pagination at the SQL level
2144 stmt = stmt.offset(skip).limit(limit)
2145 result = await session.execute(stmt)
2146 all_chats = result.scalars().all()
2148 log.info('The number of chats: %s', len(all_chats))
2150 # Validate and return chats
2151 return [ChatModel.model_validate(chat) for chat in all_chats]
2153 async def get_chats_by_folder_id_and_user_id(
2154 self,
2155 folder_id: str,
2156 user_id: str,
2157 skip: int = 0,
2158 limit: int = 60,
2159 sort_by: str = 'updated_at',
2160 sort_dir: str = 'desc',
2161 db: AsyncSession | None = None,
2162 ) -> list[ChatTitleIdResponse]:
2163 async with get_async_db_context(db) as session:
2164 stmt = (
2165 select(Chat.id, Chat.title, Chat.updated_at, Chat.created_at, Chat.last_read_at)
2166 .filter_by(folder_id=folder_id, user_id=user_id)
2167 .filter(or_(Chat.pinned == False, Chat.pinned == None))
2168 .filter_by(archived=False)
2169 .where(Chat.meta['internal'].as_boolean().is_not(True))
2170 )
2171 stmt = stmt.order_by(*chat_list_order(sort_by, sort_dir))
2173 if skip:
2174 stmt = stmt.offset(skip)
2175 if limit: 2175 ↛ 2178line 2175 didn't jump to line 2178 because the condition on line 2175 was always true
2176 stmt = stmt.limit(limit)
2178 result = await session.execute(stmt)
2179 all_chats = result.all()
2180 return [
2181 ChatTitleIdResponse.model_validate(
2182 {
2183 'id': chat[0],
2184 'title': chat[1],
2185 'updated_at': chat[2],
2186 'created_at': chat[3],
2187 'last_read_at': chat[4],
2188 }
2189 )
2190 for chat in all_chats
2191 ]
2193 async def get_all_chats_by_folder_id(
2194 self,
2195 folder_id: str,
2196 skip: int = 0,
2197 limit: int = 60,
2198 sort_by: str = 'updated_at',
2199 sort_dir: str = 'desc',
2200 unread_for_user_id: str | None = None,
2201 db: AsyncSession | None = None,
2202 ) -> list[dict]:
2203 """Get chats in a folder across ALL users. Returns dicts with user_id."""
2204 async with get_async_db_context(db) as session:
2205 stmt = (
2206 select(Chat.id, Chat.title, Chat.user_id, Chat.updated_at, Chat.created_at, Chat.last_read_at)
2207 .filter_by(folder_id=folder_id)
2208 .filter(or_(Chat.pinned == False, Chat.pinned == None))
2209 .filter_by(archived=False)
2210 .where(Chat.meta['internal'].as_boolean().is_not(True))
2211 )
2212 stmt = stmt.order_by(*chat_list_order(sort_by, sort_dir, unread_for_user_id))
2214 if skip:
2215 stmt = stmt.offset(skip)
2216 if limit: 2216 ↛ 2219line 2216 didn't jump to line 2219 because the condition on line 2216 was always true
2217 stmt = stmt.limit(limit)
2219 result = await session.execute(stmt)
2220 all_chats = result.all()
2221 return [
2222 {
2223 'id': chat[0],
2224 'title': chat[1],
2225 'user_id': chat[2],
2226 'updated_at': chat[3],
2227 'created_at': chat[4],
2228 'last_read_at': chat[5],
2229 }
2230 for chat in all_chats
2231 ]
2233 async def count_all_chats_by_folder_id(
2234 self,
2235 folder_id: str,
2236 db: AsyncSession | None = None,
2237 ) -> int:
2238 async with get_async_db_context(db) as session:
2239 stmt = (
2240 select(func.count(Chat.id))
2241 .filter_by(folder_id=folder_id)
2242 .filter(or_(Chat.pinned == False, Chat.pinned == None))
2243 .filter_by(archived=False)
2244 .where(Chat.meta['internal'].as_boolean().is_not(True))
2245 )
2246 result = await session.execute(stmt)
2247 return result.scalar_one()
2249 async def get_chats_by_folder_ids_and_user_id(
2250 self, folder_ids: list[str], user_id: str, db: AsyncSession | None = None
2251 ) -> list[ChatModel]:
2252 async with get_async_db_context(db) as session:
2253 stmt = (
2254 select(Chat)
2255 .filter(Chat.folder_id.in_(folder_ids), Chat.user_id == user_id)
2256 .filter(or_(Chat.pinned == False, Chat.pinned == None))
2257 .filter_by(archived=False)
2258 .where(Chat.meta['internal'].as_boolean().is_not(True))
2259 .order_by(Chat.updated_at.desc())
2260 )
2262 result = await session.execute(stmt)
2263 all_chats = result.scalars().all()
2264 return [ChatModel.model_validate(chat) for chat in all_chats]
2266 async def update_chat_folder_id_by_id_and_user_id(
2267 self, id: str, user_id: str, folder_id: str, db: AsyncSession | None = None
2268 ) -> ChatModel | None:
2269 try:
2270 async with get_async_db_context(db) as session:
2271 chat = await session.get(Chat, id)
2272 chat.folder_id = folder_id
2273 chat.updated_at = int(time.time())
2274 chat.last_read_at = int(time.time())
2275 chat.pinned = False
2276 if folder_id is not None:
2277 # Folder listings only show unarchived chats, so moving an archived
2278 # chat into a folder would otherwise have no visible effect: the chat
2279 # stays in the archived list and never appears in the folder.
2280 chat.archived = False
2281 await session.commit()
2282 return ChatModel.model_validate(chat)
2283 except Exception:
2284 return None
2286 async def get_chat_tags_by_id_and_user_id(
2287 self, id: str, user_id: str, db: AsyncSession | None = None
2288 ) -> list[TagModel]:
2289 async with get_async_db_context(db) as session:
2290 stmt = select(Chat.meta).where(Chat.id == id)
2291 result = await session.execute(stmt)
2292 meta = result.scalar_one_or_none()
2293 tag_ids = (meta or {}).get('tags', [])
2294 return await Tags.get_tags_by_ids_and_user_id(tag_ids, user_id, db=session)
2296 async def get_chat_list_by_user_id_and_tag_name(
2297 self,
2298 user_id: str,
2299 tag_name: str,
2300 skip: int = 0,
2301 limit: int = 50,
2302 db: AsyncSession | None = None,
2303 ) -> list[ChatTitleIdResponse]:
2304 async with get_async_db_context(db) as session:
2305 stmt = select(Chat.id, Chat.title, Chat.updated_at, Chat.created_at, Chat.last_read_at).filter_by(
2306 user_id=user_id
2307 )
2308 stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True))
2309 tag_id = tag_name.replace(' ', '_').lower()
2311 bind = await session.connection()
2312 dialect_name = bind.dialect.name
2313 log.info('DB dialect name: %s', dialect_name)
2314 if dialect_name == 'sqlite': 2314 ↛ 2318line 2314 didn't jump to line 2318 because the condition on line 2314 was always true
2315 stmt = stmt.filter(
2316 text(f"EXISTS (SELECT 1 FROM json_each(Chat.meta, '$.tags') WHERE json_each.value = :tag_id)")
2317 ).params(tag_id=tag_id)
2318 elif dialect_name == 'postgresql':
2319 stmt = stmt.filter(
2320 text("EXISTS (SELECT 1 FROM json_array_elements_text(Chat.meta->'tags') elem WHERE elem = :tag_id)")
2321 ).params(tag_id=tag_id)
2322 else:
2323 raise NotImplementedError(f'Unsupported dialect: {dialect_name}')
2325 stmt = stmt.order_by(Chat.updated_at.desc(), Chat.id)
2327 if skip:
2328 stmt = stmt.offset(skip)
2329 if limit:
2330 stmt = stmt.limit(limit)
2332 result = await session.execute(stmt)
2333 all_chats = result.all()
2334 return [
2335 ChatTitleIdResponse.model_validate(
2336 {
2337 'id': chat[0],
2338 'title': chat[1],
2339 'updated_at': chat[2],
2340 'created_at': chat[3],
2341 'last_read_at': chat[4],
2342 }
2343 )
2344 for chat in all_chats
2345 ]
2347 async def add_chat_tag_by_id_and_user_id_and_tag_name(
2348 self, id: str, user_id: str, tag_name: str, db: AsyncSession | None = None
2349 ) -> None:
2350 """Add one tag to a chat's meta. Meta-column-only, never the blob."""
2351 tag_id = tag_name.replace(' ', '_').lower()
2352 await Tags.ensure_tags_exist([tag_name], user_id, db=db)
2353 try:
2354 async with get_async_db_context(db) as session:
2355 row = (await session.execute(select(Chat.meta).filter_by(id=id))).one_or_none()
2356 if row is None:
2357 return None
2359 meta = row[0] or {}
2360 if tag_id not in meta.get('tags', []):
2361 await session.execute(
2362 update(Chat)
2363 .filter_by(id=id)
2364 .values(meta={**meta, 'tags': list(set(meta.get('tags', []) + [tag_id]))})
2365 )
2366 await session.commit()
2367 except Exception:
2368 return None
2370 async def count_chats_by_tag_name_and_user_id(
2371 self, tag_name: str, user_id: str, db: AsyncSession | None = None
2372 ) -> int:
2373 tag_id = tag_name.replace(' ', '_').lower()
2374 counts = await self.count_chats_by_tag_ids_and_user_id([tag_id], user_id, db=db)
2375 return counts.get(tag_id, 0)
2377 async def count_chats_by_tag_ids_and_user_id(
2378 self, tag_ids: list[str], user_id: str, db: AsyncSession | None = None
2379 ) -> dict[str, int]:
2380 """Per-tag chat counts in one round trip (one scalar subquery per tag)."""
2381 if not tag_ids: 2381 ↛ 2382line 2381 didn't jump to line 2382 because the condition on line 2381 was never true
2382 return {}
2383 async with get_async_db_context(db) as session:
2384 bind = await session.connection()
2385 dialect_name = bind.dialect.name
2387 columns = []
2388 for index, tag_id in enumerate(tag_ids):
2389 tag_id = tag_id.replace(' ', '_').lower()
2390 stmt = select(func.count(Chat.id)).filter_by(user_id=user_id, archived=False)
2391 stmt = stmt.where(Chat.meta['internal'].as_boolean().is_not(True))
2392 param = f'tag_id_{index}'
2393 if dialect_name == 'sqlite': 2393 ↛ 2397line 2393 didn't jump to line 2397 because the condition on line 2393 was always true
2394 stmt = stmt.filter(
2395 text(f"EXISTS (SELECT 1 FROM json_each(Chat.meta, '$.tags') WHERE json_each.value = :{param})")
2396 ).params(**{param: tag_id})
2397 elif dialect_name == 'postgresql':
2398 stmt = stmt.filter(
2399 text(
2400 f"EXISTS (SELECT 1 FROM json_array_elements_text(Chat.meta->'tags') elem WHERE elem = :{param})"
2401 )
2402 ).params(**{param: tag_id})
2403 else:
2404 raise NotImplementedError(f'Unsupported dialect: {dialect_name}')
2405 columns.append(stmt.scalar_subquery().label(f'count_{index}'))
2407 row = (await session.execute(select(*columns))).one()
2408 return dict(zip(tag_ids, row))
2410 async def delete_orphan_tags_for_user(
2411 self,
2412 tag_ids: list[str],
2413 user_id: str,
2414 threshold: int = 0,
2415 db: AsyncSession | None = None,
2416 ) -> None:
2417 """Delete tag rows from *tag_ids* that appear in at most *threshold*
2418 non-archived chats for *user_id*. One query to find orphans, one to
2419 delete them.
2421 Use threshold=0 after a tag is already removed from a chat's meta.
2422 Use threshold=1 when the chat itself is about to be deleted (the
2423 referencing chat still exists at query time).
2424 """
2425 if not tag_ids:
2426 return
2427 async with get_async_db_context(db) as session:
2428 counts = await self.count_chats_by_tag_ids_and_user_id(tag_ids, user_id, db=session)
2429 orphans = [tag_id for tag_id in tag_ids if counts.get(tag_id, 0) <= threshold]
2430 await Tags.delete_tags_by_ids_and_user_id(orphans, user_id, db=session)
2432 async def count_chats_by_folder_id_and_user_id(
2433 self, folder_id: str, user_id: str, db: AsyncSession | None = None
2434 ) -> int:
2435 async with get_async_db_context(db) as session:
2436 stmt = select(func.count(Chat.id)).filter_by(user_id=user_id, folder_id=folder_id)
2437 result = await session.execute(stmt.where(Chat.meta['internal'].as_boolean().is_not(True)))
2438 count = result.scalar()
2440 log.info("Count of chats for folder '%s': %s", folder_id, count)
2441 return count
2443 async def count_chats_by_folder_ids_and_user_id(
2444 self, folder_ids: list[str], user_id: str, db: AsyncSession | None = None
2445 ) -> int:
2446 if not folder_ids:
2447 return 0
2449 async with get_async_db_context(db) as session:
2450 stmt = select(func.count(Chat.id)).filter(Chat.user_id == user_id, Chat.folder_id.in_(folder_ids))
2451 result = await session.execute(stmt.where(Chat.meta['internal'].as_boolean().is_not(True)))
2452 count = result.scalar()
2454 log.info("Count of chats for folders '%s': %s", folder_ids, count)
2455 return count
2457 async def delete_tag_by_id_and_user_id_and_tag_name(
2458 self, id: str, user_id: str, tag_name: str, db: AsyncSession | None = None
2459 ) -> bool:
2460 try:
2461 async with get_async_db_context(db) as session:
2462 chat = await session.get(Chat, id)
2463 tags = chat.meta.get('tags', [])
2464 tag_id = tag_name.replace(' ', '_').lower()
2466 tags = [tag for tag in tags if tag != tag_id]
2467 chat.meta = {
2468 **chat.meta,
2469 'tags': list(set(tags)),
2470 }
2471 await session.commit()
2472 return True
2473 except Exception:
2474 return False
2476 async def delete_chat_by_id(self, id: str, db: AsyncSession | None = None) -> bool:
2477 try:
2478 async with get_async_db_context(db) as session:
2479 await session.execute(update(AutomationRun).filter_by(chat_id=id).values(chat_id=None))
2480 await session.execute(delete(ChatMessage).filter_by(chat_id=id))
2481 await session.execute(delete(Chat).filter_by(id=id))
2482 await session.commit()
2484 return True and await self.delete_shared_chat_by_chat_id(id, db=session)
2485 except Exception:
2486 return False
2488 async def delete_chat_by_id_and_user_id(self, id: str, user_id: str, db: AsyncSession | None = None) -> bool:
2489 try:
2490 async with get_async_db_context(db) as session:
2491 await session.execute(update(AutomationRun).filter_by(chat_id=id).values(chat_id=None))
2492 await session.execute(delete(ChatMessage).filter_by(chat_id=id))
2493 await session.execute(delete(Chat).filter_by(id=id, user_id=user_id))
2494 await session.commit()
2496 return True and await self.delete_shared_chat_by_chat_id(id, db=session)
2497 except Exception:
2498 return False
2500 async def delete_chats_by_user_id(self, user_id: str, db: AsyncSession | None = None) -> bool:
2501 try:
2502 async with get_async_db_context(db) as session:
2503 await self.delete_shared_chats_by_user_id(user_id, db=session)
2505 chat_id_subquery = select(Chat.id).filter_by(user_id=user_id).scalar_subquery()
2506 await session.execute(
2507 update(AutomationRun)
2508 .filter(AutomationRun.chat_id.in_(select(Chat.id).filter_by(user_id=user_id)))
2509 .values(chat_id=None)
2510 )
2511 await session.execute(
2512 delete(ChatMessage).filter(ChatMessage.chat_id.in_(select(Chat.id).filter_by(user_id=user_id)))
2513 )
2514 await session.execute(delete(Chat).filter_by(user_id=user_id))
2515 await session.commit()
2517 return True
2518 except Exception:
2519 return False
2521 async def delete_chats_by_user_id_and_folder_id(
2522 self, user_id: str, folder_id: str, db: AsyncSession | None = None
2523 ) -> bool:
2524 try:
2525 async with get_async_db_context(db) as session:
2526 chat_ids_stmt = select(Chat.id).filter_by(user_id=user_id, folder_id=folder_id)
2527 await session.execute(
2528 update(AutomationRun).filter(AutomationRun.chat_id.in_(chat_ids_stmt)).values(chat_id=None)
2529 )
2530 await session.execute(delete(ChatMessage).filter(ChatMessage.chat_id.in_(chat_ids_stmt)))
2531 await session.execute(delete(Chat).filter_by(user_id=user_id, folder_id=folder_id))
2532 await session.commit()
2534 return True
2535 except Exception:
2536 return False
2538 async def move_chats_by_folder_id(
2539 self,
2540 folder_id: str,
2541 new_folder_id: str | None,
2542 db: AsyncSession | None = None,
2543 ) -> bool:
2544 try:
2545 async with get_async_db_context(db) as session:
2546 await session.execute(update(Chat).filter_by(folder_id=folder_id).values(folder_id=new_folder_id))
2547 await session.commit()
2549 return True
2550 except Exception:
2551 return False
2553 async def delete_shared_chats_by_user_id(self, user_id: str, db: AsyncSession | None = None) -> bool:
2554 """Delete all shared chat snapshots created by a user."""
2555 from open_webui.models.shared_chats import SharedChat as SharedChatTable
2556 from open_webui.models.shared_chats import SharedChats
2558 try:
2559 async with get_async_db_context(db) as session:
2560 # Delete shared_chat rows for this user's chats
2561 await session.execute(delete(SharedChatTable).filter_by(user_id=user_id))
2563 # Clear share_id on all of this user's chats
2564 await session.execute(update(Chat).filter_by(user_id=user_id).values(share_id=None))
2565 await session.commit()
2567 return True
2568 except Exception:
2569 return False
2571 async def insert_chat_files(
2572 self,
2573 chat_id: str,
2574 message_id: str,
2575 file_ids: list[str],
2576 user_id: str,
2577 db: AsyncSession | None = None,
2578 ) -> list[ChatFileModel] | None:
2579 if not file_ids:
2580 return None
2582 chat_message_file_ids = {
2583 item.id for item in await self.get_chat_files_by_chat_id_and_message_id(chat_id, message_id, db=db)
2584 }
2585 # Remove duplicates and existing file_ids
2586 file_ids = list({file_id for file_id in file_ids if file_id and file_id not in chat_message_file_ids})
2587 if not file_ids:
2588 return None
2590 # Only link files the caller can read; blocks forging a chat_file row to another user's file.
2591 from open_webui.models.files import Files
2592 from open_webui.models.users import Users
2593 from open_webui.utils.access_control.files import has_access_to_file
2595 user = await Users.get_user_by_id(user_id, db=db)
2596 accessible_file_ids = []
2597 for file_id in file_ids:
2598 file = await Files.get_file_by_id(file_id, db=db)
2599 if not file:
2600 continue
2601 if ( 2601 ↛ anywhereline 2601 didn't jump anywhere: it always raised an exception.
2602 file.user_id == user_id
2603 or (user and user.role == 'admin')
2604 or (user and await has_access_to_file(file_id, 'read', user, db=db))
2605 ):
2606 accessible_file_ids.append(file_id)
2607 file_ids = accessible_file_ids
2608 if not file_ids:
2609 return None
2611 try:
2612 async with get_async_db_context(db) as session:
2613 now = int(time.time())
2615 chat_files = [
2616 ChatFileModel(
2617 id=str(uuid.uuid4()),
2618 user_id=user_id,
2619 chat_id=chat_id,
2620 message_id=message_id,
2621 file_id=file_id,
2622 created_at=now,
2623 updated_at=now,
2624 )
2625 for file_id in file_ids
2626 ]
2628 results = [ChatFile(**chat_file.model_dump()) for chat_file in chat_files]
2630 session.add_all(results)
2631 await session.commit()
2633 return chat_files
2634 except Exception:
2635 return None
2637 async def get_chat_files_by_chat_id_and_message_id(
2638 self, chat_id: str, message_id: str, db: AsyncSession | None = None
2639 ) -> list[ChatFileModel]:
2640 async with get_async_db_context(db) as session:
2641 result = await session.execute(
2642 select(ChatFile).filter_by(chat_id=chat_id, message_id=message_id).order_by(ChatFile.created_at.asc())
2643 )
2644 all_chat_files = result.scalars().all()
2645 return [ChatFileModel.model_validate(chat_file) for chat_file in all_chat_files]
2647 async def delete_chat_file(self, chat_id: str, file_id: str, db: AsyncSession | None = None) -> bool:
2648 try:
2649 async with get_async_db_context(db) as session:
2650 await session.execute(delete(ChatFile).filter_by(chat_id=chat_id, file_id=file_id))
2651 await session.commit()
2652 return True
2653 except Exception:
2654 return False
2656 async def get_shared_chat_ids_by_file_id(self, file_id: str, db: AsyncSession | None = None) -> list[str]:
2657 """Return IDs of chats that contain this file and have an active share link."""
2658 async with get_async_db_context(db) as session:
2659 result = await session.execute(
2660 select(Chat.id)
2661 .join(ChatFile, Chat.id == ChatFile.chat_id)
2662 .filter(ChatFile.file_id == file_id, Chat.share_id.isnot(None))
2663 )
2664 return [row[0] for row in result.all()]
2666 async def update_chat_tasks_by_id(self, id: str, tasks: list[dict]) -> ChatModel | None:
2667 """Update the tasks list on a chat."""
2668 try:
2669 async with get_async_db_context() as session:
2670 chat = await session.get(Chat, id)
2671 if chat is None:
2672 return None
2673 chat.tasks = tasks
2674 await session.commit()
2675 return ChatModel.model_validate(chat)
2676 except Exception:
2677 return None
2679 async def get_chat_tasks_by_id(self, id: str) -> list[dict]:
2680 """Read the tasks list from a chat (lightweight column query)."""
2681 async with get_async_db_context() as session:
2682 result = await session.execute(select(Chat.tasks).filter_by(id=id))
2683 row = result.first()
2684 if row is None or row[0] is None:
2685 return []
2686 return row[0]
2689Chats = ChatTable() # singleton chats repository