Coverage for src/local_deep_research/research_library/deletion/routes/delete_routes.py: 98%
170 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""
2Delete API Routes
4Provides endpoints for delete operations:
5- Delete document
6- Delete document blob only
7- Delete documents in bulk
8- Delete blobs in bulk
9- Remove document from collection
10- Delete collection
11- Delete collection index only
12"""
14from flask import Blueprint, jsonify, request, session
17from ....web.auth.decorators import login_required
18from ...utils import handle_api_error
19from ..services.document_deletion import DocumentDeletionService
20from ..services.collection_deletion import (
21 CollectionDeletionService,
22 PROTECTED_COLLECTION_TYPES,
23)
24from ..services.bulk_deletion import BulkDeletionService
27delete_bp = Blueprint("delete", __name__, url_prefix="/library/api")
28# NOTE: Routes use session["username"] (not .get()) intentionally.
29# @login_required guarantees the key exists; direct access fails fast
30# if the decorator is ever removed.
33class _ValidationError(ValueError):
34 """Raised when request validation fails."""
36 def __init__(self, message: str):
37 super().__init__(message)
38 self.message = message
41def _validate_document_ids() -> list:
42 """Extract and validate document_ids from the JSON request body.
44 Returns:
45 list: The validated document IDs.
47 Raises:
48 _ValidationError: If the request body is missing or invalid.
49 """
50 data = request.get_json()
51 if not data or "document_ids" not in data:
52 raise _ValidationError("document_ids required in request body")
54 document_ids = data["document_ids"]
55 if not isinstance(document_ids, list) or not document_ids:
56 raise _ValidationError("document_ids must be a non-empty list")
58 return document_ids
61# =============================================================================
62# Document Delete Endpoints
63# =============================================================================
66@delete_bp.route("/document/<string:document_id>", methods=["DELETE"])
67@login_required
68def delete_document(document_id):
69 """
70 Delete a document and all related data.
72 Tooltip: "Permanently delete this document, including PDF and text content.
73 This cannot be undone."
75 Returns:
76 JSON with deletion details including chunks deleted, blob size freed
77 """
78 try:
79 username = session["username"]
80 service = DocumentDeletionService(username)
81 result = service.delete_document(document_id)
83 if result.get("deleted"):
84 return jsonify({"success": True, **result})
85 # Note refusal (bypass guard): caller targeted a note via the
86 # document API. Return 403 so the frontend can route them to
87 # DELETE /api/notes/<id>, distinct from 404 (not found).
88 if result.get("is_note"):
89 return jsonify({"success": False, **result}), 403
90 return jsonify({"success": False, **result}), 404
92 except Exception as e:
93 return handle_api_error("deleting document", e)
96@delete_bp.route("/document/<string:document_id>/blob", methods=["DELETE"])
97@login_required
98def delete_document_blob(document_id):
99 """
100 Delete PDF binary but keep document metadata and text content.
102 Tooltip: "Remove the PDF file to save space. Text content will be
103 preserved for searching."
105 Returns:
106 JSON with bytes freed
107 """
108 try:
109 username = session["username"]
110 service = DocumentDeletionService(username)
111 result = service.delete_blob_only(document_id)
113 if result.get("deleted"):
114 return jsonify({"success": True, **result})
115 # Note refusal: mirror the sibling DELETE /document/<id> route
116 # which returns 403 when delete_document declines on a note.
117 # Pre-fix this branch only returned 400 (bad request) because
118 # the error string lacked "not found", which confuses clients
119 # routing 403 → "use the notes API instead" vs 400 → "bad
120 # request payload".
121 if result.get("is_note"):
122 return jsonify({"success": False, **result}), 403
123 error_code = (
124 404 if "not found" in result.get("error", "").lower() else 400
125 )
126 return jsonify({"success": False, **result}), error_code
128 except Exception as e:
129 return handle_api_error("deleting document blob", e)
132@delete_bp.route("/document/<string:document_id>/preview", methods=["GET"])
133@login_required
134def get_document_deletion_preview(document_id):
135 """
136 Get a preview of what will be deleted.
138 Returns information about the document to help user confirm deletion.
139 """
140 try:
141 username = session["username"]
142 service = DocumentDeletionService(username)
143 result = service.get_deletion_preview(document_id)
145 if result.get("found"): 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true
146 return jsonify({"success": True, **result})
147 return jsonify({"success": False, "error": "Document not found"}), 404
149 except Exception as e:
150 return handle_api_error("getting document preview", e)
153# =============================================================================
154# Collection Document Endpoints
155# =============================================================================
158@delete_bp.route(
159 "/collection/<string:collection_id>/document/<string:document_id>",
160 methods=["DELETE"],
161)
162@login_required
163def remove_document_from_collection(collection_id, document_id):
164 """
165 Remove document from a collection.
167 If the document is not in any other collection, it will be deleted.
169 Tooltip: "Remove from this collection. If not in any other collection,
170 the document will be deleted."
172 Returns:
173 JSON with unlink status and whether document was deleted
174 """
175 try:
176 username = session["username"]
177 service = DocumentDeletionService(username)
178 result = service.remove_from_collection(document_id, collection_id)
180 if result.get("unlinked"):
181 return jsonify({"success": True, **result})
182 # Protected-home refusal (note in its Notes collection): 403 so
183 # the frontend can distinguish it from 404 (not found / not in
184 # this collection), mirroring the is_note guard on delete_document.
185 if result.get("protected"):
186 return jsonify({"success": False, **result}), 403
187 return jsonify({"success": False, **result}), 404
189 except Exception as e:
190 return handle_api_error("removing document from collection", e)
193# =============================================================================
194# Collection Delete Endpoints
195# =============================================================================
198@delete_bp.route("/collections/<string:collection_id>", methods=["DELETE"])
199@login_required
200def delete_collection(collection_id):
201 """
202 Delete a collection and clean up all related data.
204 Deletes the collection, its RAG index, chunks, and any orphaned
205 documents (documents not in any other collection).
207 Returns:
208 JSON with deletion details including orphaned documents deleted
209 """
210 try:
211 username = session["username"]
212 service = CollectionDeletionService(username)
213 result = service.delete_collection(
214 collection_id, delete_orphaned_documents=True
215 )
217 if result.get("deleted"):
218 return jsonify({"success": True, **result})
219 error = result.get("error", "Unknown error")
220 if "not found" in error.lower():
221 status_code = 404
222 elif result.get("collection_type") in PROTECTED_COLLECTION_TYPES:
223 status_code = 409
224 else:
225 status_code = 400
226 return jsonify({"success": False, **result}), status_code
228 except Exception as e:
229 return handle_api_error("deleting collection", e)
232@delete_bp.route(
233 "/collections/<string:collection_id>/index", methods=["DELETE"]
234)
235@login_required
236def delete_collection_index(collection_id):
237 """
238 Delete only the RAG index for a collection, keeping the collection itself.
240 Useful for rebuilding an index from scratch.
242 Returns:
243 JSON with deletion details
244 """
245 try:
246 username = session["username"]
247 service = CollectionDeletionService(username)
248 result = service.delete_collection_index_only(collection_id)
250 if result.get("deleted"):
251 return jsonify({"success": True, **result})
252 return jsonify({"success": False, **result}), 404
254 except Exception as e:
255 return handle_api_error("deleting collection index", e)
258@delete_bp.route("/collections/<string:collection_id>/preview", methods=["GET"])
259@login_required
260def get_collection_deletion_preview(collection_id):
261 """
262 Get a preview of what will be deleted.
264 Returns information about the collection to help user confirm deletion.
265 """
266 try:
267 username = session["username"]
268 service = CollectionDeletionService(username)
269 result = service.get_deletion_preview(collection_id)
271 if result.get("found"): 271 ↛ 272line 271 didn't jump to line 272 because the condition on line 271 was never true
272 return jsonify({"success": True, **result})
273 return jsonify({"success": False, "error": "Collection not found"}), 404
275 except Exception as e:
276 return handle_api_error("getting collection preview", e)
279# =============================================================================
280# Bulk Delete Endpoints
281# =============================================================================
284@delete_bp.route("/documents/bulk", methods=["DELETE"])
285@login_required
286def delete_documents_bulk():
287 """
288 Delete multiple documents at once.
290 Tooltip: "Permanently delete all selected documents and their associated data."
292 Request body:
293 {"document_ids": ["id1", "id2", ...]}
295 Returns:
296 JSON with bulk deletion results
297 """
298 try:
299 document_ids = _validate_document_ids()
300 username = session["username"]
301 service = BulkDeletionService(username)
302 result = service.delete_documents(document_ids)
304 return jsonify({"success": True, **result})
306 except _ValidationError as e:
307 return jsonify({"success": False, "error": e.message}), 400
308 except Exception as e:
309 return handle_api_error("bulk deleting documents", e)
312@delete_bp.route("/documents/blobs", methods=["DELETE"])
313@login_required
314def delete_documents_blobs_bulk():
315 """
316 Delete PDF binaries for multiple documents.
318 Tooltip: "Remove PDF files from selected documents to free up database space.
319 Text content is preserved."
321 Request body:
322 {"document_ids": ["id1", "id2", ...]}
324 Returns:
325 JSON with bulk blob deletion results
326 """
327 try:
328 document_ids = _validate_document_ids()
329 username = session["username"]
330 service = BulkDeletionService(username)
331 result = service.delete_blobs(document_ids)
333 return jsonify({"success": True, **result})
335 except _ValidationError as e:
336 return jsonify({"success": False, "error": e.message}), 400
337 except Exception as e:
338 return handle_api_error("bulk deleting blobs", e)
341@delete_bp.route(
342 "/collection/<string:collection_id>/documents/bulk", methods=["DELETE"]
343)
344@login_required
345def remove_documents_from_collection_bulk(collection_id):
346 """
347 Remove multiple documents from a collection.
349 Documents that are not in any other collection will be deleted.
351 Request body:
352 {"document_ids": ["id1", "id2", ...]}
354 Returns:
355 JSON with bulk removal results
356 """
357 try:
358 document_ids = _validate_document_ids()
359 username = session["username"]
360 service = BulkDeletionService(username)
361 result = service.remove_documents_from_collection(
362 document_ids, collection_id
363 )
365 return jsonify({"success": True, **result})
367 except _ValidationError as e:
368 return jsonify({"success": False, "error": e.message}), 400
369 except Exception as e:
370 return handle_api_error("bulk removing documents from collection", e)
373@delete_bp.route("/documents/preview", methods=["POST"])
374@login_required
375def get_bulk_deletion_preview():
376 """
377 Get a preview of what will be affected by a bulk operation.
379 Request body:
380 {
381 "document_ids": ["id1", "id2", ...],
382 "operation": "delete" or "delete_blobs"
383 }
385 Returns:
386 JSON with preview information
387 """
388 try:
389 document_ids = _validate_document_ids()
390 data = request.get_json()
391 operation = data.get("operation", "delete")
393 username = session["username"]
394 service = BulkDeletionService(username)
395 result = service.get_bulk_preview(document_ids, operation)
397 return jsonify({"success": True, **result})
399 except _ValidationError as e:
400 return jsonify({"success": False, "error": e.message}), 400
401 except Exception as e:
402 return handle_api_error("getting bulk preview", e)