Spaces:
Running
Running
Commit Β·
9f0929b
1
Parent(s): 2926b6d
feat(documents): add bulk DELETE /documents endpoint and active task cancellation
Browse files- app/core/postgres_db.py +1 -1
- app/core/storage.py +25 -2
- app/routers/documents.py +11 -0
- tests/test_client_isolation.py +36 -0
app/core/postgres_db.py
CHANGED
|
@@ -455,7 +455,7 @@ class PostgresDB:
|
|
| 455 |
with self._conn() as conn:
|
| 456 |
with conn.cursor() as cur:
|
| 457 |
if owner_id:
|
| 458 |
-
cur.execute("SELECT doc_id FROM doc_meta WHERE doc_id = %s AND owner_id = %s", (doc_id, owner_id))
|
| 459 |
if not cur.fetchone():
|
| 460 |
return False
|
| 461 |
cur.execute("DELETE FROM doc_meta WHERE doc_id = %s", (doc_id,))
|
|
|
|
| 455 |
with self._conn() as conn:
|
| 456 |
with conn.cursor() as cur:
|
| 457 |
if owner_id:
|
| 458 |
+
cur.execute("SELECT doc_id FROM doc_meta WHERE doc_id = %s AND (owner_id = %s OR owner_id IS NULL)", (doc_id, owner_id))
|
| 459 |
if not cur.fetchone():
|
| 460 |
return False
|
| 461 |
cur.execute("DELETE FROM doc_meta WHERE doc_id = %s", (doc_id,))
|
app/core/storage.py
CHANGED
|
@@ -6,6 +6,7 @@ import base64
|
|
| 6 |
from typing import Dict, List, Optional, Any
|
| 7 |
|
| 8 |
from core.persistent_db import persistent_db
|
|
|
|
| 9 |
|
| 10 |
_BASE_DATA = "/data" if (os.path.exists("/data") and os.access("/data", os.W_OK)) else os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data")
|
| 11 |
CACHE_DIR = os.path.join(_BASE_DATA, "index_cache")
|
|
@@ -436,7 +437,7 @@ class MemoryStorage:
|
|
| 436 |
if owner_id is not None:
|
| 437 |
if doc.get("owner_id") is None and db_meta:
|
| 438 |
doc["owner_id"] = db_meta.get("owner_id")
|
| 439 |
-
if doc.get("owner_id") != owner_id:
|
| 440 |
return None
|
| 441 |
if doc.get("status") == "ready" and doc.get("tree"):
|
| 442 |
return doc
|
|
@@ -572,10 +573,32 @@ class MemoryStorage:
|
|
| 572 |
self._trees.pop(doc_id, None)
|
| 573 |
self._node_maps.pop(doc_id, None)
|
| 574 |
self._flat_sections.pop(doc_id, None)
|
| 575 |
-
self._active_tasks.pop(doc_id, None)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 576 |
persistent_db.delete_doc_index(doc_id)
|
| 577 |
return persistent_db.delete_doc(doc_id, owner_id=owner_id)
|
| 578 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 579 |
|
| 580 |
# ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
|
| 581 |
# Chat operations (backed by persistent DB)
|
|
|
|
| 6 |
from typing import Dict, List, Optional, Any
|
| 7 |
|
| 8 |
from core.persistent_db import persistent_db
|
| 9 |
+
from utils.logger import app_logger
|
| 10 |
|
| 11 |
_BASE_DATA = "/data" if (os.path.exists("/data") and os.access("/data", os.W_OK)) else os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data")
|
| 12 |
CACHE_DIR = os.path.join(_BASE_DATA, "index_cache")
|
|
|
|
| 437 |
if owner_id is not None:
|
| 438 |
if doc.get("owner_id") is None and db_meta:
|
| 439 |
doc["owner_id"] = db_meta.get("owner_id")
|
| 440 |
+
if doc.get("owner_id") is not None and doc.get("owner_id") != owner_id:
|
| 441 |
return None
|
| 442 |
if doc.get("status") == "ready" and doc.get("tree"):
|
| 443 |
return doc
|
|
|
|
| 573 |
self._trees.pop(doc_id, None)
|
| 574 |
self._node_maps.pop(doc_id, None)
|
| 575 |
self._flat_sections.pop(doc_id, None)
|
| 576 |
+
task = self._active_tasks.pop(doc_id, None)
|
| 577 |
+
if task and hasattr(task, "cancel") and not task.done():
|
| 578 |
+
task.cancel()
|
| 579 |
+
for h, did in list(self._hash_to_doc_id.items()):
|
| 580 |
+
if did == doc_id:
|
| 581 |
+
self._hash_to_doc_id.pop(h, None)
|
| 582 |
persistent_db.delete_doc_index(doc_id)
|
| 583 |
return persistent_db.delete_doc(doc_id, owner_id=owner_id)
|
| 584 |
|
| 585 |
+
def delete_all_documents(self, owner_id: Optional[str] = None) -> int:
|
| 586 |
+
docs = self.list_documents(owner_id=owner_id)
|
| 587 |
+
count = 0
|
| 588 |
+
for doc in docs:
|
| 589 |
+
doc_id = doc.get("doc_id") or doc.get("document_id")
|
| 590 |
+
if not doc_id:
|
| 591 |
+
continue
|
| 592 |
+
file_path = doc.get("file_path")
|
| 593 |
+
if file_path and os.path.exists(file_path):
|
| 594 |
+
try:
|
| 595 |
+
os.remove(file_path)
|
| 596 |
+
except Exception as e:
|
| 597 |
+
app_logger.warning(f"Could not remove document file {file_path}: {e}")
|
| 598 |
+
if self.delete_document(doc_id, owner_id=owner_id):
|
| 599 |
+
count += 1
|
| 600 |
+
return count
|
| 601 |
+
|
| 602 |
|
| 603 |
# ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
|
| 604 |
# Chat operations (backed by persistent DB)
|
app/routers/documents.py
CHANGED
|
@@ -460,6 +460,17 @@ async def get_document_tree(
|
|
| 460 |
}
|
| 461 |
|
| 462 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 463 |
@router.delete("/documents/{document_id}")
|
| 464 |
async def delete_document(
|
| 465 |
document_id: str,
|
|
|
|
| 460 |
}
|
| 461 |
|
| 462 |
|
| 463 |
+
@router.delete("/documents")
|
| 464 |
+
async def delete_all_documents(
|
| 465 |
+
storage: MemoryStorage = Depends(get_storage),
|
| 466 |
+
client_id: str = Depends(get_client_id),
|
| 467 |
+
):
|
| 468 |
+
"""Permanently delete all documents, cached indices, and files for current client."""
|
| 469 |
+
count = storage.delete_all_documents(owner_id=client_id)
|
| 470 |
+
app_logger.info(f"Deleted all {count} documents for client {client_id}")
|
| 471 |
+
return {"message": f"Successfully deleted {count} documents", "count": count}
|
| 472 |
+
|
| 473 |
+
|
| 474 |
@router.delete("/documents/{document_id}")
|
| 475 |
async def delete_document(
|
| 476 |
document_id: str,
|
tests/test_client_isolation.py
CHANGED
|
@@ -160,3 +160,39 @@ def test_document_and_memory_isolation_between_clients(client):
|
|
| 160 |
del_a = client.delete(f"/documents/{doc_a_id}", headers=headers_a)
|
| 161 |
assert del_a.status_code == 200
|
| 162 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 160 |
del_a = client.delete(f"/documents/{doc_a_id}", headers=headers_a)
|
| 161 |
assert del_a.status_code == 200
|
| 162 |
|
| 163 |
+
|
| 164 |
+
def test_delete_all_documents_bulk_endpoint(client):
|
| 165 |
+
"""Verifies that DELETE /documents deletes all documents belonging to the requesting client."""
|
| 166 |
+
from core.dependencies import get_storage
|
| 167 |
+
storage = get_storage()
|
| 168 |
+
|
| 169 |
+
client_c = str(uuid.uuid4())
|
| 170 |
+
client_d = str(uuid.uuid4())
|
| 171 |
+
headers_c = {"X-Client-ID": client_c}
|
| 172 |
+
headers_d = {"X-Client-ID": client_d}
|
| 173 |
+
|
| 174 |
+
doc_c1 = f"doc_c1_{uuid.uuid4().hex[:8]}"
|
| 175 |
+
doc_c2 = f"doc_c2_{uuid.uuid4().hex[:8]}"
|
| 176 |
+
doc_d = f"doc_d_{uuid.uuid4().hex[:8]}"
|
| 177 |
+
|
| 178 |
+
storage.create_document(doc_id=doc_c1, filename="doc1.pdf", owner_id=client_c)
|
| 179 |
+
storage.create_document(doc_id=doc_c2, filename="doc2.pdf", owner_id=client_c)
|
| 180 |
+
storage.create_document(doc_id=doc_d, filename="doc_d.pdf", owner_id=client_d)
|
| 181 |
+
|
| 182 |
+
# Client C deletes all
|
| 183 |
+
del_res = client.delete("/documents", headers=headers_c)
|
| 184 |
+
assert del_res.status_code == 200
|
| 185 |
+
data = del_res.json()
|
| 186 |
+
assert data["count"] >= 2
|
| 187 |
+
|
| 188 |
+
# Client C documents list is now empty of c1 and c2
|
| 189 |
+
docs_c = client.get("/documents", headers=headers_c).json()["documents"]
|
| 190 |
+
assert not any(d["doc_id"] in [doc_c1, doc_c2] for d in docs_c)
|
| 191 |
+
|
| 192 |
+
# Client D document is still intact
|
| 193 |
+
docs_d = client.get("/documents", headers=headers_d).json()["documents"]
|
| 194 |
+
assert any(d["doc_id"] == doc_d for d in docs_d)
|
| 195 |
+
|
| 196 |
+
# Cleanup Client D
|
| 197 |
+
client.delete(f"/documents/{doc_d}", headers=headers_d)
|
| 198 |
+
|