Coverage for open_webui/models/notes.py: 68%
249 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 05:07 +0000
1import time
2import uuid
3from functools import lru_cache
4from typing import Optional
6from open_webui.internal.db import Base, get_async_db_context
7from open_webui.models.access_grants import AccessGrantModel, AccessGrants
8from open_webui.models.groups import Groups
9from open_webui.models.users import User, UserModel, UserResponse, Users
10from open_webui.utils.json_codec import JSONCodec
11from pydantic import BaseModel, ConfigDict, Field, field_validator
12from sqlalchemy import JSON, BigInteger, Boolean, Column, ForeignKey, Text, delete, func, or_, select, update
13from sqlalchemy.ext.asyncio import AsyncSession
15####################
16# Note DB Schema
17####################
20class Note(Base):
21 __tablename__ = 'note'
23 id = Column(Text, primary_key=True, unique=True)
24 user_id = Column(Text)
26 title = Column(Text)
27 data = Column(JSON, nullable=True)
28 meta = Column(JSON, nullable=True)
30 created_at = Column(BigInteger)
31 updated_at = Column(BigInteger)
34def sanitize_note_data(data: Optional[dict]) -> Optional[dict]:
35 """Sanitize malformed note.data so content.md is always markdown text."""
36 if data is None:
37 return None
38 if not isinstance(data, dict):
39 return {'content': {'md': str(data)}}
41 content = data.get('content')
42 if not isinstance(content, dict) or 'md' not in content or isinstance(content.get('md'), str): 42 ↛ 45line 42 didn't jump to line 45 because the condition on line 42 was always true
43 return data
45 md = content.get('md') if content.get('md') is not None else ''
46 if isinstance(md, (dict, list)):
47 md = f'```json\n{JSONCodec.dumps(md, indent=2, ensure_ascii=False)}\n```'
48 else:
49 md = str(md)
51 return {
52 **data,
53 'content': {
54 **content,
55 'md': md,
56 },
57 }
60class NoteModel(BaseModel):
61 model_config = ConfigDict(from_attributes=True)
63 id: str
64 user_id: str
66 title: str
67 data: Optional[dict] = None
68 meta: Optional[dict] = None
69 is_pinned: Optional[bool] = False
71 access_grants: list[AccessGrantModel] = Field(default_factory=list)
73 created_at: int # timestamp in epoch
74 updated_at: int # timestamp in epoch
76 @field_validator('data', mode='before')
77 @classmethod
78 def sanitize_data(cls, data):
79 return sanitize_note_data(data)
82class PinnedNote(Base):
83 __tablename__ = 'pinned_note'
85 id = Column(Text, primary_key=True)
86 user_id = Column(Text, nullable=False)
87 note_id = Column(Text, ForeignKey('note.id', ondelete='CASCADE'), nullable=False)
88 created_at = Column(BigInteger, nullable=False)
91####################
92# Forms
93####################
96class NoteForm(BaseModel):
97 title: str
98 data: Optional[dict] = None
99 meta: Optional[dict] = None
100 access_grants: Optional[list[dict]] = None
102 @field_validator('data', mode='before')
103 @classmethod
104 def sanitize_data(cls, data):
105 return sanitize_note_data(data)
108class NoteUpdateForm(BaseModel):
109 title: Optional[str] = None
110 data: Optional[dict] = None
111 meta: Optional[dict] = None
112 access_grants: Optional[list[dict]] = None
114 @field_validator('data', mode='before')
115 @classmethod
116 def sanitize_data(cls, data):
117 return sanitize_note_data(data)
120class NoteUserResponse(NoteModel):
121 user: Optional[UserResponse] = None
124class NoteItemResponse(BaseModel):
125 id: str
126 title: str
127 data: Optional[dict]
128 is_pinned: Optional[bool] = False
129 updated_at: int
130 created_at: int
131 user: Optional[UserResponse] = None
134class NoteListResponse(BaseModel):
135 items: list[NoteUserResponse]
136 total: int
139class NoteTable:
140 async def _get_access_grants(self, note_id: str, db: Optional[AsyncSession] = None) -> list[AccessGrantModel]:
141 return await AccessGrants.get_grants_by_resource('note', note_id, db=db)
143 async def _to_note_model(
144 self,
145 note: Note,
146 access_grants: Optional[list[AccessGrantModel]] = None,
147 db: Optional[AsyncSession] = None,
148 ) -> NoteModel:
149 # We exclude access_grants to inject them
150 note_model = NoteModel.model_validate(note)
151 note_model.data = note_model.data or {}
152 note_model.access_grants = (
153 access_grants if access_grants is not None else await self._get_access_grants(note_model.id, db=db)
154 )
155 return note_model
157 def _has_permission(self, db, query, filter: dict, permission: str = 'read'):
158 return AccessGrants.has_permission_filter(
159 db=db,
160 query=query,
161 DocumentModel=Note,
162 filter=filter,
163 resource_type='note',
164 permission=permission,
165 )
167 async def insert_new_note(
168 self, user_id: str, form_data: NoteForm, db: Optional[AsyncSession] = None
169 ) -> Optional[NoteModel]:
170 async with get_async_db_context(db) as db:
171 note = NoteModel(
172 **{
173 'id': str(uuid.uuid4()),
174 'user_id': user_id,
175 **form_data.model_dump(exclude={'access_grants'}),
176 'created_at': int(time.time_ns()),
177 'updated_at': int(time.time_ns()),
178 'access_grants': [],
179 }
180 )
182 new_note = Note(**note.model_dump(exclude={'access_grants', 'is_pinned'}))
184 db.add(new_note)
185 await db.commit()
186 await AccessGrants.set_access_grants('note', note.id, form_data.access_grants, db=db)
187 return await self._to_note_model(new_note, db=db)
189 async def get_notes(self, skip: int = 0, limit: int = 50, db: Optional[AsyncSession] = None) -> list[NoteModel]:
190 async with get_async_db_context(db) as db:
191 stmt = select(Note).order_by(Note.updated_at.desc())
192 if skip is not None:
193 stmt = stmt.offset(skip)
194 if limit is not None:
195 stmt = stmt.limit(limit)
196 result = await db.execute(stmt)
197 notes = result.scalars().all()
198 note_ids = [note.id for note in notes]
199 grants_map = await AccessGrants.get_grants_by_resources('note', note_ids, db=db)
200 return [await self._to_note_model(note, access_grants=grants_map.get(note.id, []), db=db) for note in notes]
202 async def search_notes(
203 self,
204 user_id: str,
205 filter: dict = {},
206 skip: int = 0,
207 limit: int = 30,
208 db: Optional[AsyncSession] = None,
209 ) -> NoteListResponse:
210 async with get_async_db_context(db) as db:
211 stmt = select(Note, User).outerjoin(User, User.id == Note.user_id)
212 if filter:
213 query_key = filter.get('query')
214 if query_key:
215 # Split query into individual words and normalize each
216 # (strip hyphens so "todo" matches "to-do").
217 # All words must match somewhere in title OR content (AND semantics).
218 search_words = query_key.split()
219 normalized_words = [w.replace('-', '') for w in search_words if w.replace('-', '')]
220 for word in normalized_words:
221 stmt = stmt.filter(
222 or_(
223 func.replace(func.replace(Note.title, '-', ''), ' ', '').ilike(f'%{word}%'),
224 func.replace(
225 func.replace(Note.data['content']['md'].as_string(), '-', ''),
226 ' ',
227 '',
228 ).ilike(f'%{word}%'),
229 )
230 )
232 view_option = filter.get('view_option')
233 if view_option == 'created': 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true
234 stmt = stmt.filter(Note.user_id == user_id)
235 elif view_option == 'shared': 235 ↛ 236line 235 didn't jump to line 236 because the condition on line 235 was never true
236 stmt = stmt.filter(Note.user_id != user_id)
238 # Apply access control filtering
239 if 'permission' in filter:
240 permission = filter['permission']
241 else:
242 permission = 'write'
244 stmt = self._has_permission(
245 db,
246 stmt,
247 filter,
248 permission=permission,
249 )
251 order_by = filter.get('order_by')
252 direction = filter.get('direction')
254 if order_by == 'name': 254 ↛ 255line 254 didn't jump to line 255 because the condition on line 254 was never true
255 if direction == 'asc':
256 stmt = stmt.order_by(Note.title.asc())
257 else:
258 stmt = stmt.order_by(Note.title.desc())
259 elif order_by == 'created_at': 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true
260 if direction == 'asc':
261 stmt = stmt.order_by(Note.created_at.asc())
262 else:
263 stmt = stmt.order_by(Note.created_at.desc())
264 elif order_by == 'updated_at': 264 ↛ 265line 264 didn't jump to line 265 because the condition on line 264 was never true
265 if direction == 'asc':
266 stmt = stmt.order_by(Note.updated_at.asc())
267 else:
268 stmt = stmt.order_by(Note.updated_at.desc())
269 else:
270 stmt = stmt.order_by(Note.updated_at.desc())
272 else:
273 stmt = stmt.order_by(Note.updated_at.desc())
275 # Count BEFORE pagination
276 count_result = await db.execute(select(func.count()).select_from(stmt.subquery()))
277 total = count_result.scalar()
279 if skip:
280 stmt = stmt.offset(skip)
281 if limit:
282 stmt = stmt.limit(limit)
284 result = await db.execute(stmt)
285 items = result.all()
287 note_ids = [note.id for note, _ in items]
288 grants_map = await AccessGrants.get_grants_by_resources('note', note_ids, db=db)
290 notes = []
291 for note, user in items:
292 notes.append(
293 NoteUserResponse(
294 **(
295 await self._to_note_model(
296 note,
297 access_grants=grants_map.get(note.id, []),
298 db=db,
299 )
300 ).model_dump(),
301 user=(UserResponse(**UserModel.model_validate(user).model_dump()) if user else None),
302 )
303 )
305 return NoteListResponse(items=notes, total=total)
307 async def get_notes_by_user_id(
308 self,
309 user_id: str,
310 permission: str = 'read',
311 skip: int = 0,
312 limit: int = 50,
313 db: Optional[AsyncSession] = None,
314 ) -> list[NoteModel]:
315 async with get_async_db_context(db) as db:
316 user_groups = await Groups.get_groups_by_member_id(user_id, db=db)
317 user_group_ids = [group.id for group in user_groups]
319 stmt = select(Note).order_by(Note.updated_at.desc())
320 stmt = self._has_permission(db, stmt, {'user_id': user_id, 'group_ids': user_group_ids}, permission)
322 if skip is not None:
323 stmt = stmt.offset(skip)
324 if limit is not None:
325 stmt = stmt.limit(limit)
327 result = await db.execute(stmt)
328 notes = result.scalars().all()
329 note_ids = [note.id for note in notes]
330 grants_map = await AccessGrants.get_grants_by_resources('note', note_ids, db=db)
331 return [await self._to_note_model(note, access_grants=grants_map.get(note.id, []), db=db) for note in notes]
333 async def get_note_by_id(self, id: str, db: Optional[AsyncSession] = None) -> Optional[NoteModel]:
334 async with get_async_db_context(db) as db:
335 result = await db.execute(select(Note).filter(Note.id == id))
336 note = result.scalars().first()
337 return await self._to_note_model(note, db=db) if note else None
339 async def update_note_by_id(
340 self, id: str, form_data: NoteUpdateForm, db: Optional[AsyncSession] = None
341 ) -> Optional[NoteModel]:
342 async with get_async_db_context(db) as db:
343 result = await db.execute(select(Note).filter(Note.id == id))
344 note = result.scalars().first()
345 if not note:
346 return None
348 form_data = form_data.model_dump(exclude_unset=True)
349 note.data = sanitize_note_data(note.data) or {}
351 if 'title' in form_data:
352 note.title = form_data['title']
353 if 'data' in form_data:
354 note.data = {**(note.data or {}), **(form_data['data'] or {})}
355 if 'meta' in form_data:
356 note.meta = {**(note.meta or {}), **(form_data['meta'] or {})}
358 if not db.is_modified(note) and 'access_grants' not in form_data:
359 return await self._to_note_model(note, db=db)
361 if 'access_grants' in form_data:
362 await AccessGrants.set_access_grants('note', id, form_data['access_grants'], db=db)
364 note.updated_at = int(time.time_ns())
366 await db.commit()
367 return await self._to_note_model(note, db=db) if note else None
369 async def toggle_note_pinned_by_id(
370 self, id: str, user_id: str, db: Optional[AsyncSession] = None
371 ) -> Optional[NoteModel]:
372 try:
373 async with get_async_db_context(db) as db:
374 result = await db.execute(select(Note).filter(Note.id == id))
375 note = result.scalars().first()
376 if not note:
377 return None
379 # Check if already pinned
380 pin_result = await db.execute(select(PinnedNote).filter_by(user_id=user_id, note_id=id))
381 pinned_note = pin_result.scalars().first()
383 if pinned_note:
384 await db.execute(delete(PinnedNote).filter_by(user_id=user_id, note_id=id))
385 else:
386 new_pin = PinnedNote(
387 id=str(uuid.uuid4()), user_id=user_id, note_id=id, created_at=int(time.time_ns())
388 )
389 db.add(new_pin)
391 await db.commit()
392 return await self._to_note_model(note, db=db)
393 except Exception:
394 return None
396 async def get_pinned_notes_by_user_id(
397 self,
398 user_id: str,
399 permission: str = 'read',
400 db: Optional[AsyncSession] = None,
401 ) -> list[NoteModel]:
402 async with get_async_db_context(db) as db:
403 user_groups = await Groups.get_groups_by_member_id(user_id, db=db)
404 user_group_ids = [group.id for group in user_groups]
406 stmt = (
407 select(Note)
408 .join(PinnedNote, PinnedNote.note_id == Note.id)
409 .filter(PinnedNote.user_id == user_id)
410 .order_by(PinnedNote.created_at.desc())
411 )
412 stmt = self._has_permission(db, stmt, {'user_id': user_id, 'group_ids': user_group_ids}, permission)
414 result = await db.execute(stmt)
415 notes = result.scalars().all()
416 note_ids = [note.id for note in notes]
417 grants_map = await AccessGrants.get_grants_by_resources('note', note_ids, db=db)
418 return [await self._to_note_model(note, access_grants=grants_map.get(note.id, []), db=db) for note in notes]
420 async def delete_note_by_id(self, id: str, db: Optional[AsyncSession] = None) -> bool:
421 try:
422 async with get_async_db_context(db) as db:
423 await AccessGrants.revoke_all_access('note', id, db=db)
424 await db.execute(delete(PinnedNote).filter(PinnedNote.note_id == id))
425 await db.execute(delete(Note).filter(Note.id == id))
426 await db.commit()
427 return True
428 except Exception:
429 return False
431 async def get_pinned_note_ids(self, user_id: str, db: Optional[AsyncSession] = None) -> list[str]:
432 async with get_async_db_context(db) as db:
433 result = await db.execute(select(PinnedNote.note_id).filter_by(user_id=user_id))
434 return result.scalars().all()
437Notes = NoteTable()