Coverage for open_webui/models/prompt_history.py: 55%
95 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 05:07 +0000
1"""Prompt history model for version tracking."""
3import difflib
4import time
5import uuid
6from typing import Optional
8from open_webui.internal.db import Base, get_async_db_context
9from open_webui.models.users import UserResponse, Users
10from pydantic import BaseModel, ConfigDict
11from sqlalchemy import JSON, BigInteger, Column, Index, Text, delete, func, select
12from sqlalchemy.ext.asyncio import AsyncSession
14####################
15# PromptHistory DB Schema
16####################
19class PromptHistory(Base):
20 __tablename__ = 'prompt_history'
22 id = Column(Text, primary_key=True)
23 prompt_id = Column(Text, nullable=False, index=True)
24 parent_id = Column(Text, nullable=True) # Reference to parent commit
25 snapshot = Column(JSON, nullable=False)
26 user_id = Column(Text, nullable=False)
27 commit_message = Column(Text, nullable=True)
28 created_at = Column(BigInteger, nullable=False)
31class PromptHistoryModel(BaseModel):
32 id: str
33 prompt_id: str
34 parent_id: Optional[str] = None
35 snapshot: dict
36 user_id: str
37 commit_message: Optional[str] = None
38 created_at: int
40 model_config = ConfigDict(from_attributes=True)
43class PromptHistoryResponse(PromptHistoryModel):
44 """Response model with user info."""
46 user: Optional[UserResponse] = None
49class PromptHistoryTable:
50 async def create_history_entry(
51 self,
52 prompt_id: str,
53 snapshot: dict,
54 user_id: str,
55 parent_id: Optional[str] = None,
56 commit_message: Optional[str] = None,
57 db: Optional[AsyncSession] = None,
58 ) -> Optional[PromptHistoryModel]:
59 """Create a new history entry (commit) for a prompt."""
60 async with get_async_db_context(db) as db:
61 history = PromptHistory(
62 id=str(uuid.uuid4()),
63 prompt_id=prompt_id,
64 parent_id=parent_id,
65 snapshot=snapshot,
66 user_id=user_id,
67 commit_message=commit_message,
68 created_at=int(time.time()),
69 )
70 db.add(history)
71 await db.commit()
72 return PromptHistoryModel.model_validate(history)
74 async def get_history_by_prompt_id(
75 self,
76 prompt_id: str,
77 limit: int = 50,
78 offset: int = 0,
79 db: Optional[AsyncSession] = None,
80 ) -> list[PromptHistoryResponse]:
81 """Get all history entries for a prompt, ordered by created_at desc."""
82 async with get_async_db_context(db) as db:
83 result = await db.execute(
84 select(PromptHistory)
85 .filter(PromptHistory.prompt_id == prompt_id)
86 .order_by(PromptHistory.created_at.desc())
87 .offset(offset)
88 .limit(limit)
89 )
90 entries = result.scalars().all()
92 # Get user info for each entry
93 user_ids = list(set(e.user_id for e in entries))
94 users = await Users.get_users_by_user_ids(user_ids, db=db) if user_ids else []
95 users_dict = {user.id: user for user in users}
97 return [
98 PromptHistoryResponse(
99 **PromptHistoryModel.model_validate(entry).model_dump(),
100 user=(users_dict.get(entry.user_id).model_dump() if users_dict.get(entry.user_id) else None),
101 )
102 for entry in entries
103 ]
105 async def get_history_entry_by_id(
106 self,
107 history_id: str,
108 db: Optional[AsyncSession] = None,
109 ) -> Optional[PromptHistoryModel]:
110 """Get a specific history entry by ID."""
111 async with get_async_db_context(db) as db:
112 result = await db.execute(select(PromptHistory).filter(PromptHistory.id == history_id))
113 entry = result.scalars().first()
114 if entry:
115 return PromptHistoryModel.model_validate(entry)
116 return None
118 async def get_latest_history_entry(
119 self,
120 prompt_id: str,
121 db: Optional[AsyncSession] = None,
122 ) -> Optional[PromptHistoryModel]:
123 """Get the most recent history entry for a prompt."""
124 async with get_async_db_context(db) as db:
125 result = await db.execute(
126 select(PromptHistory)
127 .filter(PromptHistory.prompt_id == prompt_id)
128 .order_by(PromptHistory.created_at.desc())
129 .limit(1)
130 )
131 entry = result.scalars().first()
132 if entry:
133 return PromptHistoryModel.model_validate(entry)
134 return None
136 async def get_history_count(
137 self,
138 prompt_id: str,
139 db: Optional[AsyncSession] = None,
140 ) -> int:
141 """Get the number of history entries for a prompt."""
142 async with get_async_db_context(db) as db:
143 result = await db.execute(
144 select(func.count()).select_from(PromptHistory).filter(PromptHistory.prompt_id == prompt_id)
145 )
146 return result.scalar()
148 async def compute_diff(
149 self,
150 from_id: str,
151 to_id: str,
152 prompt_id: str,
153 db: Optional[AsyncSession] = None,
154 ) -> Optional[dict]:
155 """Compute diff between two history entries."""
156 async with get_async_db_context(db) as db:
157 # Bind both entries to the authorized prompt; an unbound id reads another prompt's snapshot.
158 result_from = await db.execute(
159 select(PromptHistory).filter(PromptHistory.id == from_id, PromptHistory.prompt_id == prompt_id)
160 )
161 from_entry = result_from.scalars().first()
162 result_to = await db.execute(
163 select(PromptHistory).filter(PromptHistory.id == to_id, PromptHistory.prompt_id == prompt_id)
164 )
165 to_entry = result_to.scalars().first()
167 if not from_entry or not to_entry:
168 return None
170 from_snapshot = from_entry.snapshot
171 to_snapshot = to_entry.snapshot
173 # Compute diff for content field
174 from_content = from_snapshot.get('content', '')
175 to_content = to_snapshot.get('content', '')
177 diff_lines = list(
178 difflib.unified_diff(
179 from_content.splitlines(keepends=True),
180 to_content.splitlines(keepends=True),
181 fromfile=f'v{from_id[:8]}',
182 tofile=f'v{to_id[:8]}',
183 lineterm='',
184 )
185 )
187 return {
188 'from_id': from_id,
189 'to_id': to_id,
190 'from_snapshot': from_snapshot,
191 'to_snapshot': to_snapshot,
192 'content_diff': diff_lines,
193 'name_changed': from_snapshot.get('name') != to_snapshot.get('name'),
194 }
196 async def delete_history_by_prompt_id(
197 self,
198 prompt_id: str,
199 db: Optional[AsyncSession] = None,
200 ) -> bool:
201 """Delete all history entries for a prompt."""
202 async with get_async_db_context(db) as db:
203 await db.execute(delete(PromptHistory).filter(PromptHistory.prompt_id == prompt_id))
204 await db.commit()
205 return True
207 async def delete_history_entry(
208 self,
209 history_id: str,
210 prompt_id: str,
211 db: Optional[AsyncSession] = None,
212 ) -> bool:
213 """Delete a history entry and reparent its children to grandparent."""
214 async with get_async_db_context(db) as db:
215 # Bind to the authorized prompt; an unbound id deletes another prompt's history.
216 result = await db.execute(select(PromptHistory).filter_by(id=history_id, prompt_id=prompt_id))
217 entry = result.scalars().first()
218 if not entry:
219 return False
221 # Find children that reference this entry as parent
222 children_result = await db.execute(select(PromptHistory).filter_by(parent_id=history_id))
223 children = children_result.scalars().all()
225 # Reparent children to grandparent
226 for child in children:
227 child.parent_id = entry.parent_id
229 await db.delete(entry)
230 await db.commit()
231 return True
234PromptHistories = PromptHistoryTable()