[NOTICKET]: add document pipeline, simplify document API

src/api/v1/document.py

@@ -1,13 +1,12 @@
 """Document management API endpoints."""
 
-from fastapi import APIRouter, Depends, HTTPException, Request, UploadFile, File, status
+from fastapi import APIRouter, Depends, HTTPException, Request, UploadFile, File
 from sqlalchemy.ext.asyncio import AsyncSession
 from src.db.postgres.connection import get_db
 from src.document.document_service import document_service
-from src.knowledge.processing_service import knowledge_processor
-from src.storage.az_blob.az_blob import blob_storage
 from src.middlewares.logging import get_logger, log_execution
 from src.middlewares.rate_limit import limiter
+from src.pipeline.document_pipeline import document_pipeline
 from pydantic import BaseModel
 from typing import List
 
@@ -57,55 +56,10 @@ async def upload_document(
 ):
     """Upload a document."""
     if not user_id:
-        raise HTTPException(
-            status_code=400,
-            detail="user_id is required"
-        )
-
-    try:
-        # Read file content
-        content = await file.read()
-        file_size = len(content)
-
-        # Get file type
-        filename = file.filename
-        file_type = filename.split('.')[-1].lower() if '.' in filename else 'txt'
-
-        if file_type not in ['pdf', 'docx', 'txt']:
-            raise HTTPException(
-                status_code=400,
-                detail="Unsupported file type. Supported: pdf, docx, txt"
-            )
-
-        # Upload to blob storage
-        blob_name = await blob_storage.upload_file(content, filename, user_id)
-
-        # Create document record
-        document = await document_service.create_document(
-            db=db,
-            user_id=user_id,
-            filename=filename,
-            blob_name=blob_name,
-            file_size=file_size,
-            file_type=file_type
-        )
+        raise HTTPException(status_code=400, detail="user_id is required")
 
-        return {
-            "status": "success",
-            "message": "Document uploaded successfully",
-            "data": {
-                "id": document.id,
-                "filename": document.filename,
-                "status": document.status
-            }
-        }
-
-    except Exception as e:
-        logger.error(f"Upload failed for user {user_id}", error=str(e))
-        raise HTTPException(
-            status_code=500,
-            detail=f"Upload failed: {str(e)}"
-        )
+    data = await document_pipeline.upload(file, user_id, db)
+    return {"status": "success", "message": "Document uploaded successfully", "data": data}
 
 
 @router.delete("/document/delete")
@@ -116,29 +70,8 @@ async def delete_document(
     db: AsyncSession = Depends(get_db)
 ):
     """Delete a document."""
-    document = await document_service.get_document(db, document_id)
-
-    if not document:
-        raise HTTPException(
-            status_code=404,
-            detail="Document not found"
-        )
-
-    if document.user_id != user_id:
-        raise HTTPException(
-            status_code=403,
-            detail="Access denied"
-        )
-
-    success = await document_service.delete_document(db, document_id)
-
-    if success:
-        return {"status": "success", "message": "Document deleted successfully"}
-    else:
-        raise HTTPException(
-            status_code=500,
-            detail="Failed to delete document"
-        )
+    await document_pipeline.delete(document_id, user_id, db)
+    return {"status": "success", "message": "Document deleted successfully"}
 
 
 @router.post("/document/process")
@@ -149,45 +82,5 @@ async def process_document(
     db: AsyncSession = Depends(get_db)
 ):
     """Process document and ingest to vector index."""
-    document = await document_service.get_document(db, document_id)
-
-    if not document:
-        raise HTTPException(
-            status_code=404,
-            detail="Document not found"
-        )
-
-    if document.user_id != user_id:
-        raise HTTPException(
-            status_code=403,
-            detail="Access denied"
-        )
-
-    try:
-        # Update status to processing
-        await document_service.update_document_status(db, document_id, "processing")
-
-        # Process document
-        chunks_count = await knowledge_processor.process_document(document, db)
-
-        # Update status to completed
-        await document_service.update_document_status(db, document_id, "completed")
-
-        return {
-            "status": "success",
-            "message": "Document processed successfully",
-            "data": {
-                "document_id": document_id,
-                "chunks_processed": chunks_count
-            }
-        }
-
-    except Exception as e:
-        logger.error(f"Processing failed for document {document_id}", error=str(e))
-        await document_service.update_document_status(
-            db, document_id, "failed", str(e)
-        )
-        raise HTTPException(
-            status_code=500,
-            detail=f"Processing failed: {str(e)}"
-        )
+    data = await document_pipeline.process(document_id, user_id, db)
+    return {"status": "success", "message": "Document processed successfully", "data": data}

src/pipeline/document_pipeline.py

@@ -0,0 +1,80 @@
+"""Document upload and processing pipeline."""
+
+from fastapi import HTTPException, UploadFile
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.document.document_service import document_service
+from src.knowledge.processing_service import knowledge_processor
+from src.middlewares.logging import get_logger
+from src.storage.az_blob.az_blob import blob_storage
+
+logger = get_logger("document_pipeline")
+
+SUPPORTED_FILE_TYPES = ["pdf", "docx", "txt"]
+
+
+class DocumentPipeline:
+    """Orchestrates the full document upload, process, and delete flows."""
+
+    async def upload(self, file: UploadFile, user_id: str, db: AsyncSession) -> dict:
+        """Validate → upload to blob → save to DB."""
+        content = await file.read()
+        file_type = file.filename.split(".")[-1].lower() if "." in file.filename else "txt"
+
+        if file_type not in SUPPORTED_FILE_TYPES:
+            raise HTTPException(
+                status_code=400,
+                detail=f"Unsupported file type. Supported: {SUPPORTED_FILE_TYPES}",
+            )
+
+        blob_name = await blob_storage.upload_file(content, file.filename, user_id)
+        document = await document_service.create_document(
+            db=db,
+            user_id=user_id,
+            filename=file.filename,
+            blob_name=blob_name,
+            file_size=len(content),
+            file_type=file_type,
+        )
+
+        logger.info(f"Uploaded document {document.id} for user {user_id}")
+        return {"id": document.id, "filename": document.filename, "status": document.status}
+
+    async def process(self, document_id: str, user_id: str, db: AsyncSession) -> dict:
+        """Validate ownership → extract text → chunk → ingest to vector store."""
+        document = await document_service.get_document(db, document_id)
+
+        if not document:
+            raise HTTPException(status_code=404, detail="Document not found")
+        if document.user_id != user_id:
+            raise HTTPException(status_code=403, detail="Access denied")
+
+        try:
+            await document_service.update_document_status(db, document_id, "processing")
+            chunks_count = await knowledge_processor.process_document(document, db)
+            await document_service.update_document_status(db, document_id, "completed")
+
+            logger.info(f"Processed document {document_id}: {chunks_count} chunks")
+            return {"document_id": document_id, "chunks_processed": chunks_count}
+
+        except Exception as e:
+            logger.error(f"Processing failed for document {document_id}", error=str(e))
+            await document_service.update_document_status(db, document_id, "failed", str(e))
+            raise HTTPException(status_code=500, detail=f"Processing failed: {str(e)}")
+
+    async def delete(self, document_id: str, user_id: str, db: AsyncSession) -> dict:
+        """Validate ownership → delete from blob and DB."""
+        document = await document_service.get_document(db, document_id)
+
+        if not document:
+            raise HTTPException(status_code=404, detail="Document not found")
+        if document.user_id != user_id:
+            raise HTTPException(status_code=403, detail="Access denied")
+
+        await document_service.delete_document(db, document_id)
+
+        logger.info(f"Deleted document {document_id} for user {user_id}")
+        return {"document_id": document_id}
+
+
+document_pipeline = DocumentPipeline()