Coverage for open_webui/routers/prompts.py: 80%

1from __future__ import annotations 

2 

3from typing import Optional 

4 

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

6from open_webui.config import BYPASS_ADMIN_ACCESS_CONTROL 

7from open_webui.constants import ERROR_MESSAGES 

8from open_webui.events import EVENTS, publish_event 

9from open_webui.internal.db import get_async_session 

10from open_webui.models.access_grants import AccessGrants 

11from open_webui.models.config import Config 

12from open_webui.models.groups import Groups 

13from open_webui.models.prompt_history import ( 

14 PromptHistories, 

15 PromptHistoryModel, 

16 PromptHistoryResponse, 

17) 

18from open_webui.models.prompts import ( 

19 PromptAccessListResponse, 

20 PromptAccessResponse, 

21 PromptForm, 

22 PromptModel, 

23 Prompts, 

24 PromptUserResponse, 

25) 

26from open_webui.utils.access_control import filter_allowed_access_grants, has_permission 

27from open_webui.utils.auth import get_admin_user, get_verified_user 

28from pydantic import BaseModel 

29from sqlalchemy.ext.asyncio import AsyncSession 

30 

31 

32class PromptVersionUpdateForm(BaseModel): 

33 version_id: str 

34 

35 

36class PromptMetadataForm(BaseModel): 

37 name: str 

38 command: str 

39 tags: list[str] | None = None 

40 

41 

42router = APIRouter() 

43 

44PAGE_ITEM_COUNT = 30 

45 

46 

47############################ 

48# GetPrompts 

49# The hardest part is knowing what to ask. Let the right 

50# question already be here when it is needed. 

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

52 

53 

54@router.get('/', response_model=list[PromptModel]) 

55async def get_prompts(user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)): 

56 if user.role == 'admin' and BYPASS_ADMIN_ACCESS_CONTROL: 56 ↛ 59line 56 didn't jump to line 59 because the condition on line 56 was always true

57 prompts = await Prompts.get_prompts(db=db) 

58 else: 

59 prompts = await Prompts.get_prompts_by_user_id(user.id, 'read', db=db) 

60 

61 return prompts 

62 

63 

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

65async def get_prompt_tags(user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session)): 

66 if user.role == 'admin' and BYPASS_ADMIN_ACCESS_CONTROL: 66 ↛ 68line 66 didn't jump to line 68 because the condition on line 66 was always true

67 return await Prompts.get_tags(db=db) 

68 return await Prompts.get_tags_by_user_id(user.id, db=db) 

69 

70 

71@router.get('/list', response_model=PromptAccessListResponse) 

72async def get_prompt_list( 

73 query: str | None = None, 

74 view_option: str | None = None, 

75 tag: str | None = None, 

76 order_by: str | None = None, 

77 direction: str | None = None, 

78 page: int | None = 1, 

79 user=Depends(get_verified_user), 

80 db: AsyncSession = Depends(get_async_session), 

81): 

82 limit = PAGE_ITEM_COUNT 

83 

84 page = max(1, page) 

85 skip = (page - 1) * limit 

86 

87 filter = {} 

88 if query: 

89 filter['query'] = query 

90 if view_option: 

91 filter['view_option'] = view_option 

92 if tag: 

93 filter['tag'] = tag 

94 if order_by: 

95 filter['order_by'] = order_by 

96 if direction: 

97 filter['direction'] = direction 

98 

99 # Pre-fetch user group IDs once - used for both filter and write_access check 

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

101 user_group_ids = {group.id for group in groups} 

102 

103 if not (user.role == 'admin' and BYPASS_ADMIN_ACCESS_CONTROL): 103 ↛ 104line 103 didn't jump to line 104 because the condition on line 103 was never true

104 if groups: 

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

106 

107 filter['user_id'] = user.id 

108 

109 result = await Prompts.search_prompts(user.id, filter=filter, skip=skip, limit=limit, db=db) 

110 

111 # Batch-fetch writable prompt IDs in a single query instead of N has_access calls 

112 prompt_ids = [prompt.id for prompt in result.items] 

113 writable_prompt_ids = await AccessGrants.get_accessible_resource_ids( 

114 user_id=user.id, 

115 resource_type='prompt', 

116 resource_ids=prompt_ids, 

117 permission='write', 

118 user_group_ids=user_group_ids, 

119 db=db, 

120 ) 

121 

122 return PromptAccessListResponse( 

123 items=[ 

124 PromptAccessResponse( 

125 **prompt.model_dump(), 

126 write_access=( 

127 (user.role == 'admin' and BYPASS_ADMIN_ACCESS_CONTROL) 

128 or user.id == prompt.user_id 

129 or prompt.id in writable_prompt_ids 

130 ), 

131 ) 

132 for prompt in result.items 

133 ], 

134 total=result.total, 

135 ) 

136 

137 

138############################ 

139# CreateNewPrompt 

140############################ 

141 

142 

143@router.post('/create', response_model=PromptModel | None) 

144async def create_new_prompt( 

145 request: Request, 

146 form_data: PromptForm, 

147 user=Depends(get_verified_user), 

148 db: AsyncSession = Depends(get_async_session), 

149): 

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

151 await has_permission( 

152 user.id, 

153 'workspace.prompts', 

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

155 db=db, 

156 ) 

157 or await has_permission( 

158 user.id, 

159 'workspace.prompts_import', 

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

161 db=db, 

162 ) 

163 ): 

164 raise HTTPException( 

165 status_code=status.HTTP_401_UNAUTHORIZED, 

166 detail=ERROR_MESSAGES.UNAUTHORIZED, 

167 ) 

168 

169 form_data.access_grants = await filter_allowed_access_grants( 

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

171 user.id, 

172 user.role, 

173 form_data.access_grants, 

174 'sharing.public_prompts', 

175 ) 

176 

177 prompt = await Prompts.get_prompt_by_command(form_data.command, db=db) 

178 if prompt is None: 

179 prompt = await Prompts.insert_new_prompt(user.id, form_data, db=db) 

180 

181 if prompt: 181 ↛ 190line 181 didn't jump to line 190 because the condition on line 181 was always true

182 await publish_event( 

183 request, 

184 EVENTS.PROMPT_CREATED, 

185 actor=user, 

186 subject_id=prompt.id, 

187 data={'name': prompt.name, 'command': prompt.command}, 

188 ) 

189 return prompt 

190 raise HTTPException( 

191 status_code=status.HTTP_400_BAD_REQUEST, 

192 detail=ERROR_MESSAGES.DEFAULT(), 

193 ) 

194 raise HTTPException( 

195 status_code=status.HTTP_400_BAD_REQUEST, 

196 detail=ERROR_MESSAGES.COMMAND_TAKEN, 

197 ) 

198 

199 

200############################ 

201# GetPromptById 

202############################ 

203 

204 

205@router.get('/id/{prompt_id}', response_model=PromptAccessResponse | None) 

206async def get_prompt_by_id( 

207 prompt_id: str, user=Depends(get_verified_user), db: AsyncSession = Depends(get_async_session) 

208): 

209 prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) 

210 

211 if prompt: 

212 if ( 212 ↛ 238line 212 didn't jump to line 238 because the condition on line 212 was always true

213 user.role == 'admin' 

214 or prompt.user_id == user.id 

215 or await AccessGrants.has_access( 

216 user_id=user.id, 

217 resource_type='prompt', 

218 resource_id=prompt.id, 

219 permission='read', 

220 db=db, 

221 ) 

222 ): 

223 return PromptAccessResponse( 

224 **prompt.model_dump(), 

225 write_access=( 

226 (user.role == 'admin' and BYPASS_ADMIN_ACCESS_CONTROL) 

227 or user.id == prompt.user_id 

228 or await AccessGrants.has_access( 

229 user_id=user.id, 

230 resource_type='prompt', 

231 resource_id=prompt.id, 

232 permission='write', 

233 db=db, 

234 ) 

235 ), 

236 ) 

237 

238 raise HTTPException( 

239 status_code=status.HTTP_404_NOT_FOUND, 

240 detail=ERROR_MESSAGES.NOT_FOUND, 

241 ) 

242 

243 

244############################ 

245# UpdatePromptById 

246############################ 

247 

248 

249@router.post('/id/{prompt_id}/update', response_model=PromptModel | None) 

250async def update_prompt_by_id( 

251 request: Request, 

252 prompt_id: str, 

253 form_data: PromptForm, 

254 user=Depends(get_verified_user), 

255 db: AsyncSession = Depends(get_async_session), 

256): 

257 """Update a prompt's content, creating a new history entry if changed.""" 

258 prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) 

259 

260 if not prompt: 

261 raise HTTPException( 

262 status_code=status.HTTP_404_NOT_FOUND, 

263 detail=ERROR_MESSAGES.NOT_FOUND, 

264 ) 

265 

266 # Is the user the original creator, in a group with write access, or an admin 

267 if ( 267 ↛ 278line 267 didn't jump to line 278 because the condition on line 267 was never true

268 prompt.user_id != user.id 

269 and not await AccessGrants.has_access( 

270 user_id=user.id, 

271 resource_type='prompt', 

272 resource_id=prompt.id, 

273 permission='write', 

274 db=db, 

275 ) 

276 and user.role != 'admin' 

277 ): 

278 raise HTTPException( 

279 status_code=status.HTTP_401_UNAUTHORIZED, 

280 detail=ERROR_MESSAGES.ACCESS_PROHIBITED, 

281 ) 

282 

283 # Check for command collision if command is being changed 

284 if form_data.command != prompt.command: 

285 existing_prompt = await Prompts.get_prompt_by_command(form_data.command, db=db) 

286 if existing_prompt and existing_prompt.id != prompt.id: 

287 raise HTTPException( 

288 status_code=status.HTTP_400_BAD_REQUEST, 

289 detail=ERROR_MESSAGES.COMMAND_TAKEN, 

290 ) 

291 

292 form_data.access_grants = await filter_allowed_access_grants( 

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

294 user.id, 

295 user.role, 

296 form_data.access_grants, 

297 'sharing.public_prompts', 

298 ) 

299 

300 # Use the ID from the found prompt 

301 updated_prompt = await Prompts.update_prompt_by_id(prompt.id, form_data, user.id, db=db) 

302 if updated_prompt: 302 ↛ 312line 302 didn't jump to line 312 because the condition on line 302 was always true

303 await publish_event( 

304 request, 

305 EVENTS.PROMPT_UPDATED, 

306 actor=user, 

307 subject_id=updated_prompt.id, 

308 data={'name': updated_prompt.name, 'command': updated_prompt.command}, 

309 ) 

310 return updated_prompt 

311 else: 

312 raise HTTPException( 

313 status_code=status.HTTP_400_BAD_REQUEST, 

314 detail=ERROR_MESSAGES.DEFAULT(), 

315 ) 

316 

317 

318############################ 

319# UpdatePromptMetadata 

320############################ 

321 

322 

323@router.post('/id/{prompt_id}/update/meta', response_model=PromptModel | None) 

324async def update_prompt_metadata( 

325 request: Request, 

326 prompt_id: str, 

327 form_data: PromptMetadataForm, 

328 user=Depends(get_verified_user), 

329 db: AsyncSession = Depends(get_async_session), 

330): 

331 """Update prompt name and command only (no history created).""" 

332 prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) 

333 

334 if not prompt: 

335 raise HTTPException( 

336 status_code=status.HTTP_404_NOT_FOUND, 

337 detail=ERROR_MESSAGES.NOT_FOUND, 

338 ) 

339 

340 if ( 340 ↛ 351line 340 didn't jump to line 351 because the condition on line 340 was never true

341 prompt.user_id != user.id 

342 and not await AccessGrants.has_access( 

343 user_id=user.id, 

344 resource_type='prompt', 

345 resource_id=prompt.id, 

346 permission='write', 

347 db=db, 

348 ) 

349 and user.role != 'admin' 

350 ): 

351 raise HTTPException( 

352 status_code=status.HTTP_401_UNAUTHORIZED, 

353 detail=ERROR_MESSAGES.ACCESS_PROHIBITED, 

354 ) 

355 

356 # Check for command collision if command is being changed 

357 if form_data.command != prompt.command: 

358 existing_prompt = await Prompts.get_prompt_by_command(form_data.command, db=db) 

359 if existing_prompt and existing_prompt.id != prompt.id: 

360 raise HTTPException( 

361 status_code=status.HTTP_400_BAD_REQUEST, 

362 detail=ERROR_MESSAGES.COMMAND_TAKEN, 

363 ) 

364 

365 updated_prompt = await Prompts.update_prompt_metadata( 

366 prompt.id, form_data.name, form_data.command, form_data.tags, db=db 

367 ) 

368 if updated_prompt: 368 ↛ 378line 368 didn't jump to line 378 because the condition on line 368 was always true

369 await publish_event( 

370 request, 

371 EVENTS.PROMPT_UPDATED, 

372 actor=user, 

373 subject_id=updated_prompt.id, 

374 data={'name': updated_prompt.name, 'command': updated_prompt.command}, 

375 ) 

376 return updated_prompt 

377 else: 

378 raise HTTPException( 

379 status_code=status.HTTP_400_BAD_REQUEST, 

380 detail=ERROR_MESSAGES.DEFAULT(), 

381 ) 

382 

383 

384@router.post('/id/{prompt_id}/update/version', response_model=PromptModel | None) 

385async def set_prompt_version( 

386 request: Request, 

387 prompt_id: str, 

388 form_data: PromptVersionUpdateForm, 

389 user=Depends(get_verified_user), 

390 db: AsyncSession = Depends(get_async_session), 

391): 

392 prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) 

393 if not prompt: 

394 raise HTTPException( 

395 status_code=status.HTTP_404_NOT_FOUND, 

396 detail=ERROR_MESSAGES.NOT_FOUND, 

397 ) 

398 

399 if ( 399 ↛ 410line 399 didn't jump to line 410 because the condition on line 399 was never true

400 prompt.user_id != user.id 

401 and not await AccessGrants.has_access( 

402 user_id=user.id, 

403 resource_type='prompt', 

404 resource_id=prompt.id, 

405 permission='write', 

406 db=db, 

407 ) 

408 and user.role != 'admin' 

409 ): 

410 raise HTTPException( 

411 status_code=status.HTTP_401_UNAUTHORIZED, 

412 detail=ERROR_MESSAGES.ACCESS_PROHIBITED, 

413 ) 

414 

415 updated_prompt = await Prompts.update_prompt_version(prompt.id, form_data.version_id, db=db) 

416 if updated_prompt: 

417 await publish_event( 

418 request, 

419 EVENTS.PROMPT_VERSION_UPDATED, 

420 actor=user, 

421 subject_id=updated_prompt.id, 

422 data={'version_id': updated_prompt.version_id}, 

423 ) 

424 return updated_prompt 

425 else: 

426 raise HTTPException( 

427 status_code=status.HTTP_400_BAD_REQUEST, 

428 detail=ERROR_MESSAGES.DEFAULT(), 

429 ) 

430 

431 

432############################ 

433# UpdatePromptAccessById 

434############################ 

435 

436 

437class PromptAccessGrantsForm(BaseModel): 

438 access_grants: list[dict] 

439 

440 

441@router.post('/id/{prompt_id}/access/update', response_model=PromptModel | None) 

442async def update_prompt_access_by_id( 

443 request: Request, 

444 prompt_id: str, 

445 form_data: PromptAccessGrantsForm, 

446 user=Depends(get_verified_user), 

447 db: AsyncSession = Depends(get_async_session), 

448): 

449 prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) 

450 if not prompt: 

451 raise HTTPException( 

452 status_code=status.HTTP_404_NOT_FOUND, 

453 detail=ERROR_MESSAGES.NOT_FOUND, 

454 ) 

455 

456 if ( 456 ↛ 467line 456 didn't jump to line 467 because the condition on line 456 was never true

457 prompt.user_id != user.id 

458 and not await AccessGrants.has_access( 

459 user_id=user.id, 

460 resource_type='prompt', 

461 resource_id=prompt.id, 

462 permission='write', 

463 db=db, 

464 ) 

465 and user.role != 'admin' 

466 ): 

467 raise HTTPException( 

468 status_code=status.HTTP_401_UNAUTHORIZED, 

469 detail=ERROR_MESSAGES.ACCESS_PROHIBITED, 

470 ) 

471 

472 form_data.access_grants = await filter_allowed_access_grants( 

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

474 user.id, 

475 user.role, 

476 form_data.access_grants, 

477 'sharing.public_prompts', 

478 ) 

479 

480 await AccessGrants.set_access_grants('prompt', prompt_id, form_data.access_grants, db=db) 

481 

482 updated_prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) 

483 await publish_event( 

484 request, 

485 EVENTS.PROMPT_ACCESS_UPDATED, 

486 actor=user, 

487 subject_id=prompt_id, 

488 data={'name': updated_prompt.name if updated_prompt else None}, 

489 ) 

490 return updated_prompt 

491 

492 

493############################ 

494# TogglePromptActiveById 

495############################ 

496 

497 

498@router.post('/id/{prompt_id}/toggle', response_model=PromptModel | None) 

499async def toggle_prompt_active( 

500 request: Request, 

501 prompt_id: str, 

502 user=Depends(get_verified_user), 

503 db: AsyncSession = Depends(get_async_session), 

504): 

505 prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) 

506 

507 if not prompt: 

508 raise HTTPException( 

509 status_code=status.HTTP_404_NOT_FOUND, 

510 detail=ERROR_MESSAGES.NOT_FOUND, 

511 ) 

512 

513 if ( 513 ↛ 524line 513 didn't jump to line 524 because the condition on line 513 was never true

514 prompt.user_id != user.id 

515 and not await AccessGrants.has_access( 

516 user_id=user.id, 

517 resource_type='prompt', 

518 resource_id=prompt.id, 

519 permission='write', 

520 db=db, 

521 ) 

522 and user.role != 'admin' 

523 ): 

524 raise HTTPException( 

525 status_code=status.HTTP_401_UNAUTHORIZED, 

526 detail=ERROR_MESSAGES.ACCESS_PROHIBITED, 

527 ) 

528 

529 result = await Prompts.toggle_prompt_active(prompt.id, db=db) 

530 if result: 530 ↛ 540line 530 didn't jump to line 540 because the condition on line 530 was always true

531 await publish_event( 

532 request, 

533 EVENTS.PROMPT_ENABLED if result.is_active else EVENTS.PROMPT_DISABLED, 

534 actor=user, 

535 subject_id=result.id, 

536 subject_type='prompt', 

537 data={'name': result.name, 'command': result.command}, 

538 ) 

539 return result 

540 raise HTTPException( 

541 status_code=status.HTTP_400_BAD_REQUEST, 

542 detail=ERROR_MESSAGES.DEFAULT(), 

543 ) 

544 

545 

546############################ 

547# DeletePromptById 

548############################ 

549 

550 

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

552async def delete_prompt_by_id( 

553 request: Request, 

554 prompt_id: str, 

555 user=Depends(get_verified_user), 

556 db: AsyncSession = Depends(get_async_session), 

557): 

558 prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) 

559 

560 if not prompt: 

561 raise HTTPException( 

562 status_code=status.HTTP_404_NOT_FOUND, 

563 detail=ERROR_MESSAGES.NOT_FOUND, 

564 ) 

565 

566 if ( 566 ↛ 577line 566 didn't jump to line 577 because the condition on line 566 was never true

567 prompt.user_id != user.id 

568 and not await AccessGrants.has_access( 

569 user_id=user.id, 

570 resource_type='prompt', 

571 resource_id=prompt.id, 

572 permission='write', 

573 db=db, 

574 ) 

575 and user.role != 'admin' 

576 ): 

577 raise HTTPException( 

578 status_code=status.HTTP_401_UNAUTHORIZED, 

579 detail=ERROR_MESSAGES.ACCESS_PROHIBITED, 

580 ) 

581 

582 result = await Prompts.delete_prompt_by_id(prompt.id, db=db) 

583 if result: 583 ↛ 591line 583 didn't jump to line 591 because the condition on line 583 was always true

584 await publish_event( 

585 request, 

586 EVENTS.PROMPT_DELETED, 

587 actor=user, 

588 subject_id=prompt.id, 

589 data={'name': prompt.name, 'command': prompt.command}, 

590 ) 

591 return result 

592 

593 

594############################ 

595# Prompt History Endpoints 

596############################ 

597 

598 

599@router.get('/id/{prompt_id}/history', response_model=list[PromptHistoryResponse]) 

600async def get_prompt_history( 

601 prompt_id: str, 

602 page: int = 0, 

603 user=Depends(get_verified_user), 

604 db: AsyncSession = Depends(get_async_session), 

605): 

606 """Get version history for a prompt.""" 

607 PAGE_SIZE = 20 

608 

609 prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) 

610 

611 if not prompt: 

612 raise HTTPException( 

613 status_code=status.HTTP_404_NOT_FOUND, 

614 detail=ERROR_MESSAGES.NOT_FOUND, 

615 ) 

616 

617 # Check read access 

618 if not ( 618 ↛ 629line 618 didn't jump to line 629 because the condition on line 618 was never true

619 user.role == 'admin' 

620 or prompt.user_id == user.id 

621 or await AccessGrants.has_access( 

622 user_id=user.id, 

623 resource_type='prompt', 

624 resource_id=prompt.id, 

625 permission='read', 

626 db=db, 

627 ) 

628 ): 

629 raise HTTPException( 

630 status_code=status.HTTP_401_UNAUTHORIZED, 

631 detail=ERROR_MESSAGES.ACCESS_PROHIBITED, 

632 ) 

633 

634 history = await PromptHistories.get_history_by_prompt_id(prompt.id, limit=PAGE_SIZE, offset=page * PAGE_SIZE, db=db) 

635 return history 

636 

637 

638@router.get('/id/{prompt_id}/history/{history_id}', response_model=PromptHistoryModel) 

639async def get_prompt_history_entry( 

640 prompt_id: str, 

641 history_id: str, 

642 user=Depends(get_verified_user), 

643 db: AsyncSession = Depends(get_async_session), 

644): 

645 """Get a specific version from history.""" 

646 prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) 

647 

648 if not prompt: 

649 raise HTTPException( 

650 status_code=status.HTTP_404_NOT_FOUND, 

651 detail=ERROR_MESSAGES.NOT_FOUND, 

652 ) 

653 

654 # Check read access 

655 if not ( 655 ↛ 666line 655 didn't jump to line 666 because the condition on line 655 was never true

656 user.role == 'admin' 

657 or prompt.user_id == user.id 

658 or await AccessGrants.has_access( 

659 user_id=user.id, 

660 resource_type='prompt', 

661 resource_id=prompt.id, 

662 permission='read', 

663 db=db, 

664 ) 

665 ): 

666 raise HTTPException( 

667 status_code=status.HTTP_401_UNAUTHORIZED, 

668 detail=ERROR_MESSAGES.ACCESS_PROHIBITED, 

669 ) 

670 

671 history_entry = await PromptHistories.get_history_entry_by_id(history_id, db=db) 

672 if not history_entry or history_entry.prompt_id != prompt.id: 672 ↛ 678line 672 didn't jump to line 678 because the condition on line 672 was always true

673 raise HTTPException( 

674 status_code=status.HTTP_404_NOT_FOUND, 

675 detail=ERROR_MESSAGES.NOT_FOUND, 

676 ) 

677 

678 return history_entry 

679 

680 

681@router.delete('/id/{prompt_id}/history/{history_id}', response_model=bool) 

682async def delete_prompt_history_entry( 

683 prompt_id: str, 

684 history_id: str, 

685 user=Depends(get_verified_user), 

686 db: AsyncSession = Depends(get_async_session), 

687): 

688 """Delete a history entry. Cannot delete the active production version.""" 

689 prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) 

690 

691 if not prompt: 

692 raise HTTPException( 

693 status_code=status.HTTP_404_NOT_FOUND, 

694 detail=ERROR_MESSAGES.NOT_FOUND, 

695 ) 

696 

697 # Check write access 

698 if not ( 698 ↛ 709line 698 didn't jump to line 709 because the condition on line 698 was never true

699 user.role == 'admin' 

700 or prompt.user_id == user.id 

701 or await AccessGrants.has_access( 

702 user_id=user.id, 

703 resource_type='prompt', 

704 resource_id=prompt.id, 

705 permission='write', 

706 db=db, 

707 ) 

708 ): 

709 raise HTTPException( 

710 status_code=status.HTTP_401_UNAUTHORIZED, 

711 detail=ERROR_MESSAGES.ACCESS_PROHIBITED, 

712 ) 

713 

714 # Cannot delete active production version 

715 if prompt.version_id == history_id: 715 ↛ 716line 715 didn't jump to line 716 because the condition on line 715 was never true

716 raise HTTPException( 

717 status_code=status.HTTP_400_BAD_REQUEST, 

718 detail='Cannot delete the active production version', 

719 ) 

720 

721 success = await PromptHistories.delete_history_entry(history_id, prompt.id, db=db) 

722 if not success: 722 ↛ 728line 722 didn't jump to line 728 because the condition on line 722 was always true

723 raise HTTPException( 

724 status_code=status.HTTP_404_NOT_FOUND, 

725 detail=ERROR_MESSAGES.NOT_FOUND, 

726 ) 

727 

728 return success 

729 

730 

731@router.get('/id/{prompt_id}/history/diff') 

732async def get_prompt_diff( 

733 prompt_id: str, 

734 from_id: str, 

735 to_id: str, 

736 user=Depends(get_verified_user), 

737 db: AsyncSession = Depends(get_async_session), 

738): 

739 """Get diff between two versions.""" 

740 prompt = await Prompts.get_prompt_by_id(prompt_id, db=db) 

741 

742 if not prompt: 

743 raise HTTPException( 

744 status_code=status.HTTP_404_NOT_FOUND, 

745 detail=ERROR_MESSAGES.NOT_FOUND, 

746 ) 

747 

748 # Check read access 

749 if not ( 

750 user.role == 'admin' 

751 or prompt.user_id == user.id 

752 or await AccessGrants.has_access( 

753 user_id=user.id, 

754 resource_type='prompt', 

755 resource_id=prompt.id, 

756 permission='read', 

757 db=db, 

758 ) 

759 ): 

760 raise HTTPException( 

761 status_code=status.HTTP_401_UNAUTHORIZED, 

762 detail=ERROR_MESSAGES.ACCESS_PROHIBITED, 

763 ) 

764 

765 diff = await PromptHistories.compute_diff(from_id, to_id, prompt.id, db=db) 

766 if not diff: 

767 raise HTTPException( 

768 status_code=status.HTTP_404_NOT_FOUND, 

769 detail=ERROR_MESSAGES.NOT_FOUND, 

770 ) 

771 

772 return diff