Mahdi Bazrafshan
3f9677f6aa
Why: - Long benchmark and process runs block HTTP clients; task #20 required async execution. Changes: - SQLite jobs table + GET /jobs and GET /jobs/{id} - POST /benchmarks?background=true and POST /documents/{id}/process?background=true return 202 + job_id - FastAPI BackgroundTasks execute work in-process; sync paths unchanged Impact: - Jobs are lost on server restart (Option A, no external queue) Co-authored-by: Cursor <cursoragent@cursor.com>
113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
"""Document and strategy API routes.
|
|
|
|
Endpoints:
|
|
POST /documents Upload a .docx / .doc / .pdf file
|
|
POST /documents/{id}/process Run chunking strategies; ?background=true enqueues a job
|
|
DELETE /documents/{id} Remove document + vectors
|
|
GET /strategies List available chunking strategies
|
|
"""
|
|
|
|
from fastapi import APIRouter, BackgroundTasks, File, Query, UploadFile
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from src.core.exceptions import DocumentProcessingError
|
|
from src.documents.models import (
|
|
DeleteResponse,
|
|
DocumentResponse,
|
|
DocumentListResponse,
|
|
ProcessRequest,
|
|
ProcessResponse,
|
|
StrategiesResponse,
|
|
StrategyInfo,
|
|
)
|
|
from src.documents.parser import SUPPORTED_SUFFIXES
|
|
from src.documents import service
|
|
from src.jobs import service as jobs_service
|
|
from src.jobs.models import JobCreatedResponse
|
|
from src.storage import sqlite as db
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/documents", response_model=DocumentListResponse)
|
|
async def list_documents(offset: int = 0, limit: int = 50):
|
|
"""List all uploaded documents with pagination."""
|
|
result = service.list_documents(offset=offset, limit=limit)
|
|
return DocumentListResponse(
|
|
items=[DocumentResponse(**doc) for doc in result["items"]],
|
|
total=result["total"],
|
|
offset=result["offset"],
|
|
limit=result["limit"],
|
|
)
|
|
|
|
|
|
@router.post("/documents", response_model=DocumentResponse, status_code=201)
|
|
async def upload_document(file: UploadFile = File(...)):
|
|
"""Upload a document. Parses it, stores the document tree in SQLite."""
|
|
if not file.filename:
|
|
raise DocumentProcessingError("No filename provided")
|
|
|
|
lower = file.filename.lower()
|
|
if not any(lower.endswith(s) for s in SUPPORTED_SUFFIXES):
|
|
raise DocumentProcessingError(
|
|
f"Only {', '.join(SUPPORTED_SUFFIXES)} files are supported"
|
|
)
|
|
|
|
content = await file.read()
|
|
if not content:
|
|
raise DocumentProcessingError("Empty file")
|
|
|
|
doc = service.upload_document(file.filename, content)
|
|
return DocumentResponse(
|
|
id=doc["id"],
|
|
filename=doc["filename"],
|
|
paragraph_count=doc.get("paragraph_count", 0),
|
|
chunk_counts=doc.get("chunk_counts", {}),
|
|
created_at=doc["created_at"],
|
|
)
|
|
|
|
|
|
@router.post("/documents/{doc_id}/process")
|
|
async def process_document(
|
|
doc_id: str,
|
|
request: ProcessRequest,
|
|
background_tasks: BackgroundTasks,
|
|
background: bool = Query(
|
|
False, description="Run in background; returns job_id (HTTP 202)"
|
|
),
|
|
):
|
|
"""Run selected chunking strategies on an uploaded document.
|
|
|
|
Pass ``background=true`` to enqueue and poll ``GET /jobs/{job_id}``.
|
|
"""
|
|
if background:
|
|
if db.get_document(doc_id) is None:
|
|
raise DocumentProcessingError(f"Document not found: {doc_id}")
|
|
job = jobs_service.enqueue_process(doc_id, request)
|
|
background_tasks.add_task(jobs_service.run_process_job, job["id"])
|
|
body = JobCreatedResponse(
|
|
job_id=job["id"],
|
|
job_type=job["job_type"],
|
|
status=job["status"],
|
|
poll_url=f"/jobs/{job['id']}",
|
|
)
|
|
return JSONResponse(status_code=202, content=body.model_dump())
|
|
|
|
return service.process_document(doc_id, request)
|
|
|
|
|
|
@router.delete("/documents/{doc_id}", response_model=DeleteResponse)
|
|
async def delete_document(doc_id: str):
|
|
"""Delete a document and all its chunk vectors from Qdrant."""
|
|
deleted = service.delete_document(doc_id)
|
|
return DeleteResponse(deleted=deleted, document_id=doc_id)
|
|
|
|
|
|
@router.get("/strategies", response_model=StrategiesResponse)
|
|
async def list_strategies():
|
|
"""List available chunking strategies with descriptions."""
|
|
strategies = service.list_strategies()
|
|
return StrategiesResponse(
|
|
strategies=[StrategyInfo(**s) for s in strategies]
|
|
)
|