Added separate storage and fixed worker, tasks, which previously got stuck on PROCESSING (NULL PROCESSOR)

backend/app/core/storage.py

@@ -0,0 +1,32 @@
+import os
+from abc import ABC, abstractmethod
+from fastapi import UploadFile
+
+class StorageProvider(ABC):
+    @abstractmethod
+    def save(self, file: UploadFile, filename: str) -> str:
+        pass
+    
+    @abstractmethod
+    def delete(self, filepath: str) -> bool:
+        pass
+
+class LocalStorageProvider(StorageProvider):
+    def __init__(self, upload_dir: str = "uploads"):
+        self.upload_dir = upload_dir
+        os.makedirs(self.upload_dir, exist_ok=True)
+
+    def save(self, file: UploadFile, filename: str) -> str:
+        filepath = os.path.join(self.upload_dir, filename)
+        with open(filepath, "wb") as buffer:
+            while chunk := file.file.read(1024 * 1024):
+                buffer.write(chunk)
+        return filepath
+    
+    def delete(self, filepath: str) -> bool:
+        if os.path.exists(filepath):
+            os.remove(filepath)
+            return True
+        return False
+    
+active_storage: StorageProvider = LocalStorageProvider()

backend/app/models/job_model.py

@@ -6,7 +6,7 @@ class Job(Base):
     __tablename__ = "jobs"
 
     job_id = Column(String, primary_key=True, index=True)
-    file_id = Column(Integer, ForeignKey("file.id"), nullable=False)
+    file_id = Column(Integer, ForeignKey("file.id", ondelete="CASCADE"), nullable=False)
     status = Column(String, default="QUEUED")
     created_at = Column(DateTime, default=lambda: datetime.now(UTC))
     started_at = Column(DateTime, nullable=True)

backend/app/services/file_delete_service.py

@@ -2,6 +2,7 @@ import os
 from sqlalchemy.orm import Session
 from fastapi import HTTPException
 from app.models.file_model import File
+from app.core.storage import active_storage
 
 def delete_file_service(db: Session, file_id: int, user_id: int):
     file = db.query(File).filter(File.id == file_id).first()
@@ -12,8 +13,7 @@ def delete_file_service(db: Session, file_id: int, user_id: int):
     if file.owner_id != user_id:
         raise HTTPException(status_code=403, detail="Not Authorized to delete this file")
     
-    if os.path.exists(file.filepath):
-        os.remove(file.filepath)
+    active_storage.delete(file.filepath)
         
     db.delete(file)
     db.commit()

backend/app/utils/file_handler.py

@@ -1,6 +1,4 @@
-from kombu.exceptions import HttpError
-from h11._abnf import status_code
-from cv2 import detail
+from app.core.storage import active_storage
 import os
 import uuid
 import magic
@@ -51,11 +49,4 @@ def generate_unique_filename(filename: str) -> str:
     return unique_name
 
 def save_file(file: UploadFile, filename: str):
-    os.makedirs(UPLOAD_DIR, exist_ok=True)
-    filepath = os.path.join(UPLOAD_DIR, filename)
-    
-    with open(filepath, "wb") as buffer:
-        while chunk := file.file.read(1024*1024): # Uploading File in Chunks (1 MB PER Chunk)
-            buffer.write(chunk)
-            
-    return filepath
+    return active_storage.save(file, filename)

backend/app/worker/tasks.py

@@ -46,6 +46,7 @@ def process_file_task(self, file_id: int):
         job.status = "COMPLETED"
         job.finished_at = datetime.now(UTC)
         logger.info(f"Successfully processed file_id: {file_id}")
+        db.commit()
 
     except SoftTimeLimitExceeded as exc:
         logger.error(f"Task for file_id {file_id} timed out. Requeuing...")
@@ -64,7 +65,12 @@ def process_file_task(self, file_id: int):
         db.commit()
         db.close()
         raise self.retry(exc=exc)
+    
+    except BaseException as exc:
+        job.status = "FAILED"
+        job.error_message = f"CRITICAL WORKER DEATH {type(exc).__name__}"
+        db.commit()
+        raise
         
     finally:
-        db.commit()
         db.close()