edit document for new pipeline

src/api/v1/document.py

@@ -1,21 +1,20 @@
 """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.document_pipeline import document_pipeline
 from pydantic import BaseModel
 from typing import List
-
+ 
 logger = get_logger("document_api")
-
+ 
 router = APIRouter(prefix="/api/v1", tags=["Documents"])
-
-
+ 
+ 
 class DocumentResponse(BaseModel):
     id: str
     filename: str
@@ -23,8 +22,8 @@ class DocumentResponse(BaseModel):
     file_size: int
     file_type: str
     created_at: str
-
-
+ 
+ 
 @router.get("/documents/{user_id}", response_model=List[DocumentResponse])
 @log_execution(logger)
 async def list_documents(
@@ -44,8 +43,8 @@ async def list_documents(
         )
         for doc in documents
     ]
-
-
+ 
+ 
 @router.post("/document/upload")
 @limiter.limit("10/minute")
 @log_execution(logger)
@@ -57,57 +56,12 @@ 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
-        )
-
-        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)}"
-        )
-
-
+        raise HTTPException(status_code=400, detail="user_id is required")
+ 
+    data = await document_pipeline.upload(file, user_id, db)
+    return {"status": "success", "message": "Document uploaded successfully", "data": data}
+ 
+ 
 @router.delete("/document/delete")
 @log_execution(logger)
 async def delete_document(
@@ -116,31 +70,10 @@ 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")
 @log_execution(logger)
 async def process_document(
@@ -149,45 +82,6 @@ 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}
+