create queue

main.py

@@ -6,7 +6,9 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Optional
 
-from fastapi import BackgroundTasks, FastAPI, File, Form, UploadFile
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI, File, Form, UploadFile
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel, Field
@@ -16,22 +18,30 @@ from services import job_store
 from services.cv_chunker import chunk_cv
 from services.cv_converter import CVConverter
 from services.job_matcher import JobInput, JobMatcher
-from services.workers import CvWorker
+from services.workers import CvWorker, QueueManager, CvTask
 
 # ---------------------------------------------------------------------------
 # App setup
 # ---------------------------------------------------------------------------
 
+converter = CVConverter()
+matcher = JobMatcher()  # SentenceTransformer loaded once at startup
+cv_worker = CvWorker(converter)
+queue_manager = QueueManager(cv_worker, concurrency=2)
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    await queue_manager.start()
+    yield
+    await queue_manager.stop()
+
 app = FastAPI(
     title="Job Processor API",
     description="CV parsing, chunking, and job-match prediction service.",
     version="2.0.0",
+    lifespan=lifespan,
 )
 
-converter = CVConverter()
-matcher = JobMatcher()  # SentenceTransformer loaded once at startup
-cv_worker = CvWorker(converter)
-
 
 # ---------------------------------------------------------------------------
 # Shared response model
@@ -170,7 +180,6 @@ async def process_cv(file: UploadFile = File(...)):
 
 @app.post("/process-cv-async", response_model=APIResponse)
 async def process_cv_async(
-    background_tasks: BackgroundTasks,
     file: UploadFile = File(...),
     job_id: str = Form(...),
     callback_url: str = Form(...),
@@ -179,7 +188,7 @@ async def process_cv_async(
     """
     Async entry point called by the Java async worker.
     Accepts the CV file + job_id + callback_url as multipart form fields.
-    Returns 202 immediately; processes the CV in a background task.
+    Returns 202 immediately; processes the CV in a queued background task.
     """
     if not file.filename:
         return APIResponse(message="No file uploaded", statusCode=400)
@@ -196,14 +205,14 @@ async def process_cv_async(
 
     job_store.create_job(job_id)
 
-    background_tasks.add_task(
-        cv_worker.run_cv_processing,
-        job_id,
-        file_bytes,
-        filename,
-        callback_url,
-        callback_secret,
+    task = CvTask(
+        job_id=job_id,
+        file_bytes=file_bytes,
+        filename=filename,
+        callback_url=callback_url,
+        callback_secret=callback_secret
     )
+    await queue_manager.enqueue(task)
 
     return APIResponse(
         message="CV processing started",

services/workers.py

@@ -6,6 +6,7 @@ import httpx
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Optional
+from dataclasses import dataclass
 
 from services import job_store
 from services.cv_chunker import chunk_cv
@@ -111,3 +112,56 @@ class CvWorker:
         except Exception as exc:
             # Log but don't crash — Java will see job stay in PROCESSING and can time out
             print(f"[CV-ASYNC] Failed to POST callback to {callback_url}: {exc}")
+
+@dataclass
+class CvTask:
+    job_id: str
+    file_bytes: bytes
+    filename: str
+    callback_url: str
+    callback_secret: str = None
+
+class QueueManager:
+    """Manages an asyncio Queue with bounded concurrent workers."""
+    
+    def __init__(self, worker: CvWorker, concurrency: int = 2):
+        self.worker = worker
+        self.concurrency = concurrency
+        self.queue = asyncio.Queue()
+        self.tasks = []
+
+    async def start(self):
+        """Starts the background worker tasks."""
+        for _ in range(self.concurrency):
+            task = asyncio.create_task(self._worker_loop())
+            self.tasks.append(task)
+        print(f"[QUEUE] Started {self.concurrency} concurrent workers.")
+
+    async def stop(self):
+        """Cancels all running workers."""
+        for task in self.tasks:
+            task.cancel()
+        await asyncio.gather(*self.tasks, return_exceptions=True)
+        print("[QUEUE] Stopped all workers.")
+
+    async def _worker_loop(self):
+        """Continuously pulls tasks from the queue and processes them."""
+        while True:
+            try:
+                task = await self.queue.get()
+                await self.worker.run_cv_processing(
+                    task.job_id,
+                    task.file_bytes,
+                    task.filename,
+                    task.callback_url,
+                    task.callback_secret
+                )
+                self.queue.task_done()
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                print(f"[QUEUE] Unhandled error in worker loop: {e}")
+
+    async def enqueue(self, task: CvTask):
+        """Adds a new task to the queue."""
+        await self.queue.put(task)