Coverage for open_webui/routers/notes.py: 78%

1import logging

2from typing import Optional

3from uuid import uuid4

5from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, status

6from open_webui.config import (

7 BYPASS_ADMIN_ACCESS_CONTROL,

8 ENABLE_ADMIN_CHAT_ACCESS,

9 ENABLE_ADMIN_EXPORT,

10)

11from open_webui.constants import ERROR_MESSAGES

12from open_webui.events import EVENTS, publish_event

13from open_webui.internal.db import get_async_session

14from open_webui.models.access_grants import AccessGrants

15from open_webui.models.chats import ChatForm, ChatResponse, Chats

16from open_webui.models.config import Config

17from open_webui.models.groups import Groups

18from open_webui.models.notes import (

19 NoteForm,

20 NoteListResponse,

21 NoteModel,

22 Notes,

23 NoteUserResponse,

24)

25from open_webui.models.users import UserResponse, Users

26from open_webui.socket.main import sio

27from open_webui.utils.access_control import (

28 filter_allowed_access_grants,

29 has_permission,

30 has_public_read_access_grant,

31 has_public_write_access_grant,

32)

33from open_webui.utils.auth import get_admin_user, get_verified_user

34from pydantic import BaseModel

35from sqlalchemy.ext.asyncio import AsyncSession

37log = logging.getLogger(__name__)

39router = APIRouter()

42def _truncate_note_data(data: Optional[dict], max_length: int = 1000) -> Optional[dict]:

43 if not data:

44 return data

45 md = (data.get('content') or {}).get('md') or ''

46 return {'content': {'md': md[:max_length]}}

49############################

50# GetNotes

51############################

54class NoteItemResponse(BaseModel):

55 id: str

56 title: str

57 data: Optional[dict]

58 is_pinned: Optional[bool] = False

59 updated_at: int

60 created_at: int

61 user: Optional[UserResponse] = None

64@router.get('/', response_model=list[NoteItemResponse])

65async def get_notes(

66 request: Request,

67 page: Optional[int] = None,

68 user=Depends(get_verified_user),

69 db: AsyncSession = Depends(get_async_session),

70):

71 if user.role != 'admin' and not await has_permission( 71 ↛ 74line 71 didn't jump to line 74 because the condition on line 71 was never true

72 user.id, 'features.notes', await Config.get('user.permissions'), db=db

73 ):

74 raise HTTPException(

75 status_code=status.HTTP_401_UNAUTHORIZED,

76 detail=ERROR_MESSAGES.UNAUTHORIZED,

77 )

79 limit = None

80 skip = None

81 if page is not None:

82 limit = 60

83 skip = (page - 1) * limit

85 notes = await Notes.get_notes_by_user_id(user.id, 'read', skip=skip, limit=limit, db=db)

86 if not notes:

87 return []

89 user_ids = list(set(note.user_id for note in notes))

90 users = {user.id: user for user in await Users.get_users_by_user_ids(user_ids, db=db)}

92 pinned_note_ids = await Notes.get_pinned_note_ids(user.id, db=db)

94 return [

95 NoteUserResponse(

96 **{

97 **note.model_dump(),

98 'is_pinned': note.id in pinned_note_ids,

99 'data': _truncate_note_data(note.data),

100 'user': UserResponse(**users[note.user_id].model_dump()),

101 }

102 )

103 for note in notes

104 if note.user_id in users

105 ]

106

107

108############################

109# GetPinnedNotes

110############################

111

112

113@router.get('/pinned', response_model=list[NoteItemResponse])

114async def get_pinned_notes(

115 request: Request,

116 user=Depends(get_verified_user),

117 db: AsyncSession = Depends(get_async_session),

118):

119 if user.role != 'admin' and not await has_permission( 119 ↛ 122line 119 didn't jump to line 122 because the condition on line 119 was never true

120 user.id, 'features.notes', await Config.get('user.permissions'), db=db

121 ):

122 raise HTTPException(

123 status_code=status.HTTP_401_UNAUTHORIZED,

124 detail=ERROR_MESSAGES.UNAUTHORIZED,

125 )

126

127 notes = await Notes.get_pinned_notes_by_user_id(user.id, 'read', db=db)

128 if not notes:

129 return []

130

131 user_ids = list(set(note.user_id for note in notes))

132 users = {user.id: user for user in await Users.get_users_by_user_ids(user_ids, db=db)}

133

134 return [

135 NoteUserResponse(

136 **{

137 **note.model_dump(),

138 'is_pinned': True,

139 'data': _truncate_note_data(note.data),

140 'user': UserResponse(**users[note.user_id].model_dump()),

141 }

142 )

143 for note in notes

144 if note.user_id in users

145 ]

146

147

148@router.get('/search', response_model=NoteListResponse)

149async def search_notes(

150 request: Request,

151 query: Optional[str] = None,

152 view_option: Optional[str] = None,

153 permission: Optional[str] = None,

154 order_by: Optional[str] = None,

155 direction: Optional[str] = None,

156 page: Optional[int] = 1,

157 user=Depends(get_verified_user),

158 db: AsyncSession = Depends(get_async_session),

159):

160 if user.role != 'admin' and not await has_permission( 160 ↛ 163line 160 didn't jump to line 163 because the condition on line 160 was never true

161 user.id, 'features.notes', await Config.get('user.permissions'), db=db

162 ):

163 raise HTTPException(

164 status_code=status.HTTP_401_UNAUTHORIZED,

165 detail=ERROR_MESSAGES.UNAUTHORIZED,

166 )

167

168 limit = None

169 skip = None

170 if page is not None: 170 ↛ 174line 170 didn't jump to line 174 because the condition on line 170 was always true

171 limit = 60

172 skip = (page - 1) * limit

173

174 filter = {}

175 if query:

176 filter['query'] = query

177 if view_option:

178 filter['view_option'] = view_option

179 if permission:

180 filter['permission'] = permission

181 if order_by:

182 filter['order_by'] = order_by

183 if direction:

184 filter['direction'] = direction

185

186 if not user.role == 'admin' or not BYPASS_ADMIN_ACCESS_CONTROL: 186 ↛ 187line 186 didn't jump to line 187 because the condition on line 186 was never true

187 groups = await Groups.get_groups_by_member_id(user.id, db=db)

188 if groups:

189 filter['group_ids'] = [group.id for group in groups]

190

191 filter['user_id'] = user.id

192

193 result = await Notes.search_notes(user.id, filter, skip=skip, limit=limit, db=db)

194 pinned_note_ids = await Notes.get_pinned_note_ids(user.id, db=db)

195 for note in result.items:

196 note.is_pinned = note.id in pinned_note_ids

197 note.data = _truncate_note_data(note.data)

198 return result

199

200

201############################

202# CreateNewNote

203############################

204

205

206@router.post('/create', response_model=Optional[NoteModel])

207async def create_new_note(

208 request: Request,

209 form_data: NoteForm,

210 user=Depends(get_verified_user),

211 db: AsyncSession = Depends(get_async_session),

212):

213 if user.role != 'admin' and not await has_permission( 213 ↛ 216line 213 didn't jump to line 216 because the condition on line 213 was never true

214 user.id, 'features.notes', await Config.get('user.permissions'), db=db

215 ):

216 raise HTTPException(

217 status_code=status.HTTP_401_UNAUTHORIZED,

218 detail=ERROR_MESSAGES.UNAUTHORIZED,

219 )

220

221 form_data.access_grants = await filter_allowed_access_grants(

222 await Config.get('user.permissions'),

223 user.id,

224 user.role,

225 form_data.access_grants,

226 'sharing.public_notes',

227 db=db,

228 )

229

230 try:

231 note = await Notes.insert_new_note(user.id, form_data, db=db)

232 await publish_event(

233 request,

234 EVENTS.NOTE_CREATED,

235 actor=user,

236 subject_id=note.id,

237 data={'title': note.title},

238 )

239 return note

240 except Exception as e:

241 log.exception(e)

242 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())

243

244

245############################

246# GetNoteById

247############################

248

249

250class NoteResponse(NoteModel):

251 write_access: bool = False

252

253

254@router.get('/{id}', response_model=Optional[NoteResponse])

255async def get_note_by_id(

256 request: Request,

257 id: str,

258 user=Depends(get_verified_user),

259 db: AsyncSession = Depends(get_async_session),

260):

261 if user.role != 'admin' and not await has_permission( 261 ↛ 264line 261 didn't jump to line 264 because the condition on line 261 was never true

262 user.id, 'features.notes', await Config.get('user.permissions'), db=db

263 ):

264 raise HTTPException(

265 status_code=status.HTTP_401_UNAUTHORIZED,

266 detail=ERROR_MESSAGES.UNAUTHORIZED,

267 )

268

269 note = await Notes.get_note_by_id(id, db=db)

270 if not note:

271 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

272

273 if user.role != 'admin' and ( 273 ↛ 285line 273 didn't jump to line 285 because the condition on line 273 was never true

274 user.id != note.user_id

275 and (

276 not await AccessGrants.has_access(

277 user_id=user.id,

278 resource_type='note',

279 resource_id=note.id,

280 permission='read',

281 db=db,

282 )

283 )

284 ):

285 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

286

287 write_access = (

288 user.role == 'admin'

289 or (user.id == note.user_id)

290 or await AccessGrants.has_access(

291 user_id=user.id,

292 resource_type='note',

293 resource_id=note.id,

294 permission='write',

295 db=db,

296 )

297 or has_public_write_access_grant(note.access_grants)

298 )

299

300 pinned_note_ids = await Notes.get_pinned_note_ids(user.id, db=db)

301 return NoteResponse(

302 **{**note.model_dump(), 'is_pinned': note.id in pinned_note_ids},

303 write_access=write_access,

304 )

305

306

307@router.get('/{id}/chat', response_model=ChatResponse)

308async def get_note_chat_by_id(

309 request: Request,

310 id: str,

311 user=Depends(get_verified_user),

312 db: AsyncSession = Depends(get_async_session),

313):

314 log.info('[note-chat] get-or-create requested note_id=%s user_id=%s', id, user.id)

315 if user.role != 'admin' and not await has_permission( 315 ↛ 318line 315 didn't jump to line 318 because the condition on line 315 was never true

316 user.id, 'features.notes', await Config.get('user.permissions'), db=db

317 ):

318 raise HTTPException(

319 status_code=status.HTTP_401_UNAUTHORIZED,

320 detail=ERROR_MESSAGES.UNAUTHORIZED,

321 )

322

323 note = await Notes.get_note_by_id(id, db=db)

324 if not note:

325 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

326

327 if user.role != 'admin' and ( 327 ↛ 337line 327 didn't jump to line 337 because the condition on line 327 was never true

328 user.id != note.user_id

329 and not await AccessGrants.has_access(

330 user_id=user.id,

331 resource_type='note',

332 resource_id=note.id,

333 permission='read',

334 db=db,

335 )

336 ):

337 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

338

339 chat = await Chats.get_internal_chat_by_note_id(note.id, user.id, db=db)

340 if chat:

341 log.info('[note-chat] reusing hidden chat note_id=%s chat_id=%s user_id=%s', note.id, chat.id, user.id)

342 params = {**((chat.chat or {}).get('params') or {})}

343 changed = False

344 if 'note_id' in params: 344 ↛ 345line 344 didn't jump to line 345 because the condition on line 344 was never true

345 del params['note_id']

346 changed = True

347

348 system = (

349 f'CONTEXT:\nCurrent note id: {note.id}\n'

350 'This chat is attached to the current note.\n'

351 'For edit requests like make this concise, rewrite, enhance, shorten, or update: call view_note then replace_note_content.\n'

352 'Do not say an edit is done unless replace_note_content succeeds.'

353 )

354 if params.get('system') != system: 354 ↛ 355line 354 didn't jump to line 355 because the condition on line 354 was never true

355 params['system'] = system

356 changed = True

357

358 if changed: 358 ↛ 359line 358 didn't jump to line 359 because the condition on line 358 was never true

359 updated_chat = await Chats.update_chat_by_id(chat.id, {'params': params}, db=db, touch=False)

360 if updated_chat:

361 return updated_chat

362

363 return chat

364

365 chat_id = str(uuid4())

366 chat = await Chats.insert_new_chat(

367 chat_id,

368 user.id,

369 ChatForm(

370 chat={

371 'id': chat_id,

372 'title': 'Chat',

373 'models': [''],

374 'params': {

375 'system': (

376 f'CONTEXT:\nCurrent note id: {note.id}\n'

377 'This chat is attached to the current note.\n'

378 'For edit requests like make this concise, rewrite, enhance, shorten, or update: call view_note then replace_note_content.\n'

379 'Do not say an edit is done unless replace_note_content succeeds.'

380 )

381 },

382 'history': {'messages': {}, 'currentId': None},

383 'messages': [],

384 'tags': [],

385 }

386 ),

387 db=db,

388 internal_meta={'internal': True, 'type': 'note', 'note_id': note.id},

389 )

390 if not chat: 390 ↛ 391line 390 didn't jump to line 391 because the condition on line 390 was never true

391 log.error('[note-chat] failed creating hidden chat note_id=%s user_id=%s', note.id, user.id)

392 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())

393

394 log.info('[note-chat] created hidden chat note_id=%s chat_id=%s user_id=%s', note.id, chat.id, user.id)

395 return chat

396

397

398@router.get('/{id}/chats', response_model=list[ChatResponse])

399async def get_note_chats_by_id(

400 request: Request,

401 id: str,

402 user=Depends(get_verified_user),

403 db: AsyncSession = Depends(get_async_session),

404):

405 if user.role != 'admin' and not await has_permission( 405 ↛ 408line 405 didn't jump to line 408 because the condition on line 405 was never true

406 user.id, 'features.notes', await Config.get('user.permissions'), db=db

407 ):

408 raise HTTPException(

409 status_code=status.HTTP_401_UNAUTHORIZED,

410 detail=ERROR_MESSAGES.UNAUTHORIZED,

411 )

412

413 note = await Notes.get_note_by_id(id, db=db)

414 if not note:

415 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

416

417 if user.role != 'admin' and ( 417 ↛ 427line 417 didn't jump to line 427 because the condition on line 417 was never true

418 user.id != note.user_id

419 and not await AccessGrants.has_access(

420 user_id=user.id,

421 resource_type='note',

422 resource_id=note.id,

423 permission='read',

424 db=db,

425 )

426 ):

427 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

428

429 chats = await Chats.get_internal_chats_by_note_id(note.id, user.id, db=db)

430 normalized_chats = []

431 for chat in chats:

432 params = {**((chat.chat or {}).get('params') or {})}

433 changed = False

434 if 'note_id' in params: 434 ↛ 435line 434 didn't jump to line 435 because the condition on line 434 was never true

435 del params['note_id']

436 changed = True

437

438 system = (

439 f'CONTEXT:\nCurrent note id: {note.id}\n'

440 'This chat is attached to the current note.\n'

441 'For edit requests like make this concise, rewrite, enhance, shorten, or update: call view_note then replace_note_content.\n'

442 'Do not say an edit is done unless replace_note_content succeeds.'

443 )

444 if params.get('system') != system: 444 ↛ 445line 444 didn't jump to line 445 because the condition on line 444 was never true

445 params['system'] = system

446 changed = True

447

448 if changed: 448 ↛ 449line 448 didn't jump to line 449 because the condition on line 448 was never true

449 chat = await Chats.update_chat_by_id(chat.id, {'params': params}, db=db, touch=False) or chat

450

451 normalized_chats.append(chat)

452

453 return normalized_chats

454

455

456@router.post('/{id}/chat', response_model=ChatResponse)

457async def create_note_chat_by_id(

458 request: Request,

459 id: str,

460 user=Depends(get_verified_user),

461 db: AsyncSession = Depends(get_async_session),

462):

463 if user.role != 'admin' and not await has_permission( 463 ↛ 466line 463 didn't jump to line 466 because the condition on line 463 was never true

464 user.id, 'features.notes', await Config.get('user.permissions'), db=db

465 ):

466 raise HTTPException(

467 status_code=status.HTTP_401_UNAUTHORIZED,

468 detail=ERROR_MESSAGES.UNAUTHORIZED,

469 )

470

471 note = await Notes.get_note_by_id(id, db=db)

472 if not note:

473 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

474

475 if user.role != 'admin' and ( 475 ↛ 485line 475 didn't jump to line 485 because the condition on line 475 was never true

476 user.id != note.user_id

477 and not await AccessGrants.has_access(

478 user_id=user.id,

479 resource_type='note',

480 resource_id=note.id,

481 permission='read',

482 db=db,

483 )

484 ):

485 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

486

487 chat_id = str(uuid4())

488 chat = await Chats.insert_new_chat(

489 chat_id,

490 user.id,

491 ChatForm(

492 chat={

493 'id': chat_id,

494 'title': 'Chat',

495 'models': [''],

496 'params': {

497 'system': (

498 f'CONTEXT:\nCurrent note id: {note.id}\n'

499 'This chat is attached to the current note.\n'

500 'For edit requests like make this concise, rewrite, enhance, shorten, or update: call view_note then replace_note_content.\n'

501 'Do not say an edit is done unless replace_note_content succeeds.'

502 )

503 },

504 'history': {'messages': {}, 'currentId': None},

505 'messages': [],

506 'tags': [],

507 }

508 ),

509 db=db,

510 internal_meta={'internal': True, 'type': 'note', 'note_id': note.id},

511 )

512 if not chat: 512 ↛ 513line 512 didn't jump to line 513 because the condition on line 512 was never true

513 log.error('[note-chat] failed creating hidden chat note_id=%s user_id=%s', note.id, user.id)

514 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())

515

516 log.info('[note-chat] created hidden chat note_id=%s chat_id=%s user_id=%s', note.id, chat.id, user.id)

517 return chat

518

519

520############################

521# UpdateNoteById

522############################

523

524

525@router.post('/{id}/update', response_model=Optional[NoteModel])

526async def update_note_by_id(

527 request: Request,

528 id: str,

529 form_data: NoteForm,

530 user=Depends(get_verified_user),

531 db: AsyncSession = Depends(get_async_session),

532):

533 if user.role != 'admin' and not await has_permission( 533 ↛ 536line 533 didn't jump to line 536 because the condition on line 533 was never true

534 user.id, 'features.notes', await Config.get('user.permissions'), db=db

535 ):

536 raise HTTPException(

537 status_code=status.HTTP_401_UNAUTHORIZED,

538 detail=ERROR_MESSAGES.UNAUTHORIZED,

539 )

540

541 note = await Notes.get_note_by_id(id, db=db)

542 if not note:

543 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

544

545 if user.role != 'admin' and ( 545 ↛ 555line 545 didn't jump to line 555 because the condition on line 545 was never true

546 user.id != note.user_id

547 and not await AccessGrants.has_access(

548 user_id=user.id,

549 resource_type='note',

550 resource_id=note.id,

551 permission='write',

552 db=db,

553 )

554 ):

555 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

556

557 if form_data.access_grants is not None:

558 form_data.access_grants = await filter_allowed_access_grants(

559 await Config.get('user.permissions'),

560 user.id,

561 user.role,

562 form_data.access_grants,

563 'sharing.public_notes',

564 db=db,

565 )

566

567 try:

568 note = await Notes.update_note_by_id(id, form_data, db=db)

569 pinned_note_ids = await Notes.get_pinned_note_ids(user.id, db=db)

570 note.is_pinned = note.id in pinned_note_ids

571

572 event_data = note.model_dump()

573 if form_data.data is not None:

574 event_data['data'] = {

575 key: note.data.get(key) for key in form_data.data.keys() if note.data is not None and key in note.data

576 }

577

578 await sio.emit(

579 'events:note',

580 event_data,

581 to=f'note:{note.id}',

582 )

583

584 await publish_event(

585 request,

586 EVENTS.NOTE_UPDATED,

587 actor=user,

588 subject_id=note.id,

589 data={'title': note.title},

590 )

591 return note

592 except Exception as e:

593 log.exception(e)

594 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())

595

596

597############################

598# UpdateNoteAccessById

599############################

600

601

602class NoteAccessGrantsForm(BaseModel):

603 access_grants: list[dict]

604

605

606@router.post('/{id}/access/update', response_model=Optional[NoteModel])

607async def update_note_access_by_id(

608 request: Request,

609 id: str,

610 form_data: NoteAccessGrantsForm,

611 user=Depends(get_verified_user),

612 db: AsyncSession = Depends(get_async_session),

613):

614 if user.role != 'admin' and not await has_permission( 614 ↛ 617line 614 didn't jump to line 617 because the condition on line 614 was never true

615 user.id, 'features.notes', await Config.get('user.permissions'), db=db

616 ):

617 raise HTTPException(

618 status_code=status.HTTP_401_UNAUTHORIZED,

619 detail=ERROR_MESSAGES.UNAUTHORIZED,

620 )

621

622 note = await Notes.get_note_by_id(id, db=db)

623 if not note:

624 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

625

626 if user.role != 'admin' and ( 626 ↛ 636line 626 didn't jump to line 636 because the condition on line 626 was never true

627 user.id != note.user_id

628 and not await AccessGrants.has_access(

629 user_id=user.id,

630 resource_type='note',

631 resource_id=note.id,

632 permission='write',

633 db=db,

634 )

635 ):

636 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

637

638 form_data.access_grants = await filter_allowed_access_grants(

639 await Config.get('user.permissions'),

640 user.id,

641 user.role,

642 form_data.access_grants,

643 'sharing.public_notes',

644 )

645

646 await AccessGrants.set_access_grants('note', id, form_data.access_grants, db=db)

647

648 note = await Notes.get_note_by_id(id, db=db)

649 pinned_note_ids = await Notes.get_pinned_note_ids(user.id, db=db)

650 note.is_pinned = note.id in pinned_note_ids

651 await publish_event(

652 request,

653 EVENTS.NOTE_ACCESS_UPDATED,

654 actor=user,

655 subject_id=note.id,

656 )

657 return note

658

659

660############################

661# PinNoteById

662############################

663

664

665@router.post('/{id}/pin', response_model=Optional[NoteModel])

666async def pin_note_by_id(

667 request: Request,

668 id: str,

669 user=Depends(get_verified_user),

670 db: AsyncSession = Depends(get_async_session),

671):

672 if user.role != 'admin' and not await has_permission( 672 ↛ 675line 672 didn't jump to line 675 because the condition on line 672 was never true

673 user.id, 'features.notes', await Config.get('user.permissions'), db=db

674 ):

675 raise HTTPException(

676 status_code=status.HTTP_401_UNAUTHORIZED,

677 detail=ERROR_MESSAGES.UNAUTHORIZED,

678 )

679

680 note = await Notes.get_note_by_id(id, db=db)

681 if not note:

682 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

683

684 if user.role != 'admin' and ( 684 ↛ 694line 684 didn't jump to line 694 because the condition on line 684 was never true

685 user.id != note.user_id

686 and not await AccessGrants.has_access(

687 user_id=user.id,

688 resource_type='note',

689 resource_id=note.id,

690 permission='read',

691 db=db,

692 )

693 ):

694 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

695

696 note = await Notes.toggle_note_pinned_by_id(id, user.id, db=db)

697 pinned_note_ids = await Notes.get_pinned_note_ids(user.id, db=db)

698 note.is_pinned = note.id in pinned_note_ids

699 await publish_event(

700 request,

701 EVENTS.NOTE_PINNED if note.is_pinned else EVENTS.NOTE_UNPINNED,

702 actor=user,

703 subject_id=note.id,

704 subject_type='note',

705 )

706 return note

707

708

709############################

710# DeleteNoteById

711############################

712

713

714@router.delete('/{id}/delete', response_model=bool)

715async def delete_note_by_id(

716 request: Request,

717 id: str,

718 user=Depends(get_verified_user),

719 db: AsyncSession = Depends(get_async_session),

720):

721 if user.role != 'admin' and not await has_permission( 721 ↛ 724line 721 didn't jump to line 724 because the condition on line 721 was never true

722 user.id, 'features.notes', await Config.get('user.permissions'), db=db

723 ):

724 raise HTTPException(

725 status_code=status.HTTP_401_UNAUTHORIZED,

726 detail=ERROR_MESSAGES.UNAUTHORIZED,

727 )

728

729 note = await Notes.get_note_by_id(id, db=db)

730 if not note:

731 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=ERROR_MESSAGES.NOT_FOUND)

732

733 if user.role != 'admin' and ( 733 ↛ 743line 733 didn't jump to line 743 because the condition on line 733 was never true

734 user.id != note.user_id

735 and not await AccessGrants.has_access(

736 user_id=user.id,

737 resource_type='note',

738 resource_id=note.id,

739 permission='write',

740 db=db,

741 )

742 ):

743 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=ERROR_MESSAGES.DEFAULT())

744

745 try:

746 note = await Notes.delete_note_by_id(id, db=db)

747 await publish_event(

748 request,

749 EVENTS.NOTE_DELETED,

750 actor=user,

751 subject_id=id,

752 )

753 return True

754 except Exception as e:

755 log.exception(e)

756 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=ERROR_MESSAGES.DEFAULT())