Coverage for src/local_deep_research/research_library/deletion/utils/cascade_helper.py: 96%
149 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""
2Cascade helper for deletion operations.
4Handles cleanup of related records that don't have proper FK constraints:
5- DocumentChunk (source_id has no FK constraint)
6- FAISS index files
7- Filesystem files
8"""
10import time
11from pathlib import Path
12from typing import Dict, List, Optional, Any
14from loguru import logger
15from sqlalchemy.orm import Session
17from ....constants import FILE_PATH_SENTINELS
18from ....database.models.library import (
19 Document,
20 DocumentBlob,
21 DocumentChunk,
22 DocumentCollection,
23 RAGIndex,
24)
25from ....database.models.download_tracker import DownloadTracker
27# A <name>.*.tmp younger than this is assumed to be a concurrent writer's live
28# persist() temp (not yet os.replace()d into place); deleting it would crash
29# that write. Matches FaissVectorStore.persist()'s own stale-temp sweep so both
30# paths agree on when a temp is certainly crash-orphaned rather than in-flight.
31_STALE_TMP_AGE_SECONDS = 3600
34class CascadeHelper:
35 """Helper class for cleaning up related records during deletion."""
37 @staticmethod
38 def delete_document_chunks(
39 session: Session,
40 document_id: str,
41 collection_name: Optional[str] = None,
42 ) -> int:
43 """
44 Delete DocumentChunks for a document.
46 Since DocumentChunk.source_id has no FK constraint, we must manually
47 clean up chunks when deleting a document.
49 Args:
50 session: Database session
51 document_id: The document ID to delete chunks for
52 collection_name: Optional collection name to limit deletion scope
54 Returns:
55 Number of chunks deleted
56 """
57 query = session.query(DocumentChunk).filter(
58 DocumentChunk.source_id == document_id,
59 DocumentChunk.source_type == "document",
60 )
62 if collection_name:
63 query = query.filter(
64 DocumentChunk.collection_name == collection_name
65 )
67 count = query.delete(synchronize_session=False)
68 logger.debug(
69 f"Deleted {count} chunks for document {document_id[:8]}..."
70 + (f" in collection {collection_name}" if collection_name else "")
71 )
72 return count
74 @staticmethod
75 def delete_collection_chunks(
76 session: Session,
77 collection_name: str,
78 ) -> int:
79 """
80 Delete all DocumentChunks for a collection.
82 Args:
83 session: Database session
84 collection_name: The collection name (e.g., "collection_<uuid>")
86 Returns:
87 Number of chunks deleted
88 """
89 count = (
90 session.query(DocumentChunk)
91 .filter_by(collection_name=collection_name)
92 .delete(synchronize_session=False)
93 )
94 logger.debug(f"Deleted {count} chunks for collection {collection_name}")
95 return count
97 @staticmethod
98 def get_document_blob_size(session: Session, document_id: str) -> int:
99 """
100 Get the size of a document's blob in bytes.
102 Args:
103 session: Database session
104 document_id: The document ID
106 Returns:
107 Size in bytes, or 0 if no blob exists
108 """
109 blob = (
110 session.query(DocumentBlob)
111 .filter_by(document_id=document_id)
112 .first()
113 )
114 if blob and blob.pdf_binary:
115 return len(blob.pdf_binary)
116 return 0
118 @staticmethod
119 def delete_document_blob(session: Session, document_id: str) -> int:
120 """
121 Delete a document's blob record.
123 Note: This is typically handled by CASCADE, but can be called explicitly
124 for blob-only deletion.
126 Args:
127 session: Database session
128 document_id: The document ID
130 Returns:
131 Size of deleted blob in bytes
132 """
133 blob = (
134 session.query(DocumentBlob)
135 .filter_by(document_id=document_id)
136 .first()
137 )
138 if blob:
139 size = len(blob.pdf_binary) if blob.pdf_binary else 0
140 session.delete(blob)
141 logger.debug(
142 f"Deleted blob for document {document_id[:8]}... ({size} bytes)"
143 )
144 return size
145 return 0
147 @staticmethod
148 def delete_filesystem_file(file_path: Optional[str]) -> bool:
149 """
150 Delete a file from the filesystem.
152 Args:
153 file_path: Path to the file (can be relative or absolute)
155 Returns:
156 True if file was deleted, False otherwise
157 """
158 if not file_path:
159 return False
161 # Skip special path markers
162 if file_path in FILE_PATH_SENTINELS:
163 return False
165 try:
166 path = Path(file_path)
167 if path.is_file():
168 path.unlink()
169 logger.debug(f"Deleted filesystem file: {file_path}")
170 return True
171 except Exception:
172 logger.exception(f"Failed to delete filesystem file: {file_path}")
173 return False
175 @staticmethod
176 def delete_faiss_index_files(index_path: Optional[str]) -> bool:
177 """
178 Delete FAISS index files.
180 FAISS stores indices as .faiss and .pkl files.
182 Args:
183 index_path: Path to the FAISS index file (without extension)
185 Returns:
186 True if files were deleted, False otherwise
187 """
188 if not index_path:
189 return False
191 try:
192 path = Path(index_path)
193 deleted_any = False
195 # FAISS index file
196 faiss_file = path.with_suffix(".faiss")
197 if faiss_file.is_file():
198 faiss_file.unlink()
199 logger.debug(f"Deleted FAISS index file: {faiss_file}")
200 deleted_any = True
202 # Pickle file for metadata (legacy pre-cutover format)
203 pkl_file = path.with_suffix(".pkl")
204 if pkl_file.is_file():
205 pkl_file.unlink()
206 logger.debug(f"Deleted FAISS pkl file: {pkl_file}")
207 deleted_any = True
209 # Migration sidecar (.idmap.json) — the text-free position->uuid map
210 # phase-1 writes and phase-2 consumes. If a collection is deleted
211 # during the .pkl -> .idmap.json cutover window (phase-1 ran, phase-2
212 # hasn't), the sidecar would otherwise be orphaned on disk forever.
213 idmap_file = path.with_suffix(".idmap.json")
214 if idmap_file.is_file():
215 idmap_file.unlink()
216 logger.debug(f"Deleted FAISS idmap sidecar: {idmap_file}")
217 deleted_any = True
219 # Sweep crash-orphaned temp files (persist writes <name>.*.tmp) and
220 # quarantined siblings (<name>.corrupt-<ns>) for the whole family,
221 # so deleting a collection doesn't leave them on disk forever — a
222 # leftover .pkl.corrupt-* is quarantined PLAINTEXT.
223 parent = path.parent
224 now = time.time()
225 for base in (faiss_file, pkl_file, idmap_file):
226 # .tmp files are AGE-GATED: FaissVectorStore.persist() writes to
227 # a <name>.*.tmp then os.replace()s it into place, so a live temp
228 # is a concurrent writer's not-yet-renamed file — unlinking it
229 # would crash that persist with FileNotFoundError. Only sweep
230 # temps old enough to be certainly crash-orphaned (mirrors
231 # persist()'s own _STALE_TMP_AGE_SECONDS sweep). A younger temp is
232 # left; a later delete/persist sweeps it once it ages out.
233 for stray in parent.glob(f"{base.name}.*.tmp"):
234 try:
235 if now - stray.stat().st_mtime < _STALE_TMP_AGE_SECONDS:
236 continue
237 stray.unlink()
238 logger.debug(f"Removed stray temp index file: {stray}")
239 deleted_any = True
240 except FileNotFoundError:
241 # Raced a concurrent writer's os.replace() — already gone.
242 continue
243 except OSError:
244 logger.warning(
245 f"Could not remove stray index file {stray}"
246 )
247 # .corrupt-* are quarantine artefacts (and .pkl.corrupt-* is
248 # PLAINTEXT) — always safe and required to remove, no age gate.
249 for stray in parent.glob(f"{base.name}.corrupt-*"):
250 try:
251 stray.unlink()
252 logger.debug(f"Removed quarantined index file: {stray}")
253 deleted_any = True
254 except FileNotFoundError:
255 continue
256 except OSError:
257 logger.warning(
258 f"Could not remove stray index file {stray}"
259 )
261 return deleted_any
262 except Exception:
263 logger.exception(f"Failed to delete FAISS files for: {index_path}")
264 return False
266 @staticmethod
267 def delete_rag_indices_for_collection(
268 session: Session,
269 collection_name: str,
270 *,
271 unlink_files: bool = True,
272 ) -> Dict[str, Any]:
273 """
274 Delete RAGIndex records (and, by default, their FAISS files) for a
275 collection.
277 Args:
278 session: Database session
279 collection_name: The collection name (e.g., "collection_<uuid>")
280 unlink_files: When True (default) the on-disk .faiss/.pkl/.idmap
281 files are unlinked here. When False, ONLY the RAGIndex DB rows
282 are staged for deletion and the file paths are returned under
283 ``index_paths`` — the caller is responsible for unlinking them
284 AFTER its transaction commits. Deleting files in-transaction is
285 unsafe when more DB work follows before the commit: a later
286 rollback restores the RAGIndex rows but cannot restore the
287 already-unlinked files, leaving the collection pointing at a
288 missing index (silently-broken search).
290 Returns:
291 Dict with deletion results (includes ``index_paths`` when
292 ``unlink_files`` is False).
293 """
294 indices = (
295 session.query(RAGIndex)
296 .filter_by(collection_name=collection_name)
297 .all()
298 )
300 deleted_indices = 0
301 deleted_files = 0
302 index_paths: List[str] = []
304 for index in indices:
305 path = str(index.index_path)
306 if unlink_files:
307 if CascadeHelper.delete_faiss_index_files(path):
308 deleted_files += 1
309 else:
310 index_paths.append(path)
312 session.delete(index)
313 deleted_indices += 1
315 logger.debug(
316 f"Deleted {deleted_indices} RAGIndex records and {deleted_files} "
317 f"FAISS files for collection {collection_name}"
318 )
320 return {
321 "deleted_indices": deleted_indices,
322 "deleted_files": deleted_files,
323 "index_paths": index_paths,
324 }
326 @staticmethod
327 def update_download_tracker(
328 session: Session,
329 document: Document,
330 ) -> bool:
331 """
332 Update DownloadTracker when a document is deleted.
334 The FK has SET NULL, but we also need to update is_downloaded flag.
336 Args:
337 session: Database session
338 document: The document being deleted
340 Returns:
341 True if tracker was updated
342 """
343 if not document.original_url:
344 return False
346 # Get URL hash using the same method as library_service
347 from ...utils import get_url_hash
349 try:
350 url_hash = get_url_hash(str(document.original_url))
351 tracker = (
352 session.query(DownloadTracker)
353 .filter_by(url_hash=url_hash)
354 .first()
355 )
357 if tracker:
358 tracker.is_downloaded = False # type: ignore[assignment]
359 tracker.file_path = None # type: ignore[assignment]
360 logger.debug(
361 f"Updated DownloadTracker for document {document.id[:8]}..."
362 )
363 return True
364 except Exception:
365 logger.exception("Failed to update DownloadTracker")
366 return False
368 @staticmethod
369 def count_document_in_collections(
370 session: Session,
371 document_id: str,
372 ) -> int:
373 """
374 Count how many collections a document is in.
376 Args:
377 session: Database session
378 document_id: The document ID
380 Returns:
381 Number of collections the document is in
382 """
383 return (
384 session.query(DocumentCollection)
385 .filter_by(document_id=document_id)
386 .count()
387 )
389 @staticmethod
390 def get_document_collections(
391 session: Session,
392 document_id: str,
393 ) -> List[str]:
394 """
395 Get all collection IDs a document belongs to.
397 Args:
398 session: Database session
399 document_id: The document ID
401 Returns:
402 List of collection IDs
403 """
404 doc_collections = (
405 session.query(DocumentCollection.collection_id)
406 .filter_by(document_id=document_id)
407 .all()
408 )
409 return [dc.collection_id for dc in doc_collections]
411 @staticmethod
412 def delete_document_completely(
413 session: Session,
414 document_id: str,
415 ) -> bool:
416 """
417 Delete a document and all related records using query-based deletes.
419 This avoids ORM cascade issues where SQLAlchemy tries to set
420 DocumentBlob.document_id to NULL (which fails because it's a PK).
422 Deletes in order:
423 1. DocumentBlob
424 2. DocumentCollection links
425 3. Document itself
427 Note: DocumentChunks should be deleted separately before calling this,
428 as they may need collection-specific handling.
430 Args:
431 session: Database session
432 document_id: The document ID to delete
434 Returns:
435 True if document was deleted
436 """
437 # Delete blob (has document_id as PK, can't be nulled by cascade)
438 session.query(DocumentBlob).filter_by(document_id=document_id).delete(
439 synchronize_session=False
440 )
442 # Delete collection links
443 session.query(DocumentCollection).filter_by(
444 document_id=document_id
445 ).delete(synchronize_session=False)
447 # Delete document itself
448 deleted = (
449 session.query(Document)
450 .filter_by(id=document_id)
451 .delete(synchronize_session=False)
452 )
454 if deleted:
455 logger.debug(f"Deleted document {document_id[:8]}... completely")
457 return deleted > 0