mod

app.py

@@ -48,7 +48,7 @@ async def lifespan(app: FastAPI):
     logger.info("Database initialized successfully")
     
     # Start background job worker
-    from services.job_worker import start_worker, stop_worker
+    from services.gemini_job_worker import start_worker, stop_worker
     await start_worker()
     logger.info("Background job worker started")
     

routers/gemini.py

@@ -75,7 +75,7 @@ async def create_job(
     input_data: dict
 ) -> GeminiJob:
     """Create a new job in the queue with auto-assigned priority."""
-    from services.job_worker import get_priority_for_job_type
+    from services.gemini_job_worker import get_priority_for_job_type
     
     job_id = f"job_{uuid.uuid4().hex[:16]}"
     priority = get_priority_for_job_type(job_type)

services/gemini_job_worker.py

@@ -0,0 +1,218 @@
+"""
+Gemini Job Worker - Specific implementation using the modular PriorityWorkerPool.
+
+This file shows how to use the modular PriorityWorkerPool with Gemini-specific
+job processing logic.
+"""
+import logging
+from datetime import datetime, timedelta
+from typing import Optional
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from core.database import DATABASE_URL
+from core.models import GeminiJob
+from services.priority_worker_pool import (
+    PriorityWorkerPool,
+    JobProcessor,
+    WorkerConfig,
+    get_interval_for_priority
+)
+from services.gemini_service import GeminiService
+
+logger = logging.getLogger(__name__)
+
+# Job type to priority mapping for Gemini jobs
+JOB_PRIORITY_MAP = {
+    "text": "fast",
+    "analyze": "fast",
+    "animation_prompt": "fast",
+    "image": "medium",
+    "edit_image": "medium",
+    "video": "slow"
+}
+
+
+def get_priority_for_job_type(job_type: str) -> str:
+    """Get the priority tier for a Gemini job type."""
+    return JOB_PRIORITY_MAP.get(job_type, "fast")
+
+
+class GeminiJobProcessor(JobProcessor[GeminiJob]):
+    """Processes Gemini AI jobs (text, image, video generation)."""
+    
+    def __init__(self, api_key: Optional[str] = None):
+        """Initialize with optional API key (uses env var if not provided)."""
+        self.api_key = api_key
+    
+    def _get_service(self) -> GeminiService:
+        """Get a GeminiService instance."""
+        return GeminiService(api_key=self.api_key)
+    
+    async def process(self, job: GeminiJob, session: AsyncSession) -> GeminiJob:
+        """Start processing a new job."""
+        service = self._get_service()
+        input_data = job.input_data or {}
+        
+        try:
+            if job.job_type == "video":
+                job = await self._start_video(job, session, service, input_data)
+            elif job.job_type == "image":
+                job = await self._process_image(job, service, input_data)
+            elif job.job_type == "text":
+                job = await self._process_text(job, service, input_data)
+            elif job.job_type == "analyze":
+                job = await self._process_analyze(job, service, input_data)
+            elif job.job_type == "animation_prompt":
+                job = await self._process_animation_prompt(job, service, input_data)
+            else:
+                job.status = "failed"
+                job.error_message = f"Unknown job type: {job.job_type}"
+                job.completed_at = datetime.utcnow()
+        except Exception as e:
+            logger.error(f"Error processing job {job.job_id}: {e}")
+            job.status = "failed"
+            job.error_message = str(e)
+            job.completed_at = datetime.utcnow()
+        
+        return job
+    
+    async def check_status(self, job: GeminiJob, session: AsyncSession) -> GeminiJob:
+        """Check status of an in-progress job (video generation)."""
+        if job.job_type != "video" or not job.third_party_id:
+            # Non-video jobs or missing third_party_id - shouldn't happen
+            job.status = "failed"
+            job.error_message = "Invalid job state for status check"
+            job.completed_at = datetime.utcnow()
+            return job
+        
+        service = self._get_service()
+        
+        try:
+            status_result = await service.check_video_status(job.third_party_id)
+            
+            if status_result.get("done"):
+                if status_result.get("status") == "completed":
+                    video_url = status_result.get("video_url")
+                    if video_url:
+                        filename = await service.download_video(video_url, job.job_id)
+                        job.status = "completed"
+                        job.output_data = {"filename": filename}
+                    else:
+                        job.status = "failed"
+                        job.error_message = "No video URL returned"
+                else:
+                    job.status = "failed"
+                    job.error_message = status_result.get("error", "Unknown error")
+                
+                job.completed_at = datetime.utcnow()
+            else:
+                # Not done - reschedule
+                job.retry_count += 1
+                config = WorkerConfig.from_env()
+                interval = get_interval_for_priority(job.priority, config)
+                job.next_process_at = datetime.utcnow() + timedelta(seconds=interval)
+                logger.debug(f"Job {job.job_id}: retry #{job.retry_count}, next check at {job.next_process_at}")
+        
+        except Exception as e:
+            logger.error(f"Error checking video status for {job.job_id}: {e}")
+            job.retry_count += 1
+            config = WorkerConfig.from_env()
+            interval = get_interval_for_priority(job.priority, config)
+            job.next_process_at = datetime.utcnow() + timedelta(seconds=interval)
+        
+        return job
+    
+    async def _start_video(self, job: GeminiJob, session: AsyncSession, service: GeminiService, input_data: dict) -> GeminiJob:
+        """Start async video generation."""
+        result = await service.start_video_generation(
+            base64_image=input_data.get("base64_image", ""),
+            mime_type=input_data.get("mime_type", "image/jpeg"),
+            prompt=input_data.get("prompt", ""),
+            aspect_ratio=input_data.get("aspect_ratio", "16:9"),
+            resolution=input_data.get("resolution", "720p"),
+            number_of_videos=input_data.get("number_of_videos", 1)
+        )
+        job.third_party_id = result.get("gemini_operation_name")
+        
+        # Schedule first status check
+        config = WorkerConfig.from_env()
+        interval = get_interval_for_priority(job.priority, config)
+        job.next_process_at = datetime.utcnow() + timedelta(seconds=interval)
+        
+        return job
+    
+    async def _process_image(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
+        """Process image edit (synchronous)."""
+        result = await service.edit_image(
+            base64_image=input_data.get("base64_image", ""),
+            mime_type=input_data.get("mime_type", "image/jpeg"),
+            prompt=input_data.get("prompt", "")
+        )
+        job.status = "completed"
+        job.output_data = {"image": result}
+        job.completed_at = datetime.utcnow()
+        return job
+    
+    async def _process_text(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
+        """Process text generation (synchronous)."""
+        result = await service.generate_text(
+            prompt=input_data.get("prompt", ""),
+            model=input_data.get("model")
+        )
+        job.status = "completed"
+        job.output_data = {"text": result}
+        job.completed_at = datetime.utcnow()
+        return job
+    
+    async def _process_analyze(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
+        """Process image analysis (synchronous)."""
+        result = await service.analyze_image(
+            base64_image=input_data.get("base64_image", ""),
+            mime_type=input_data.get("mime_type", "image/jpeg"),
+            prompt=input_data.get("prompt", "")
+        )
+        job.status = "completed"
+        job.output_data = {"analysis": result}
+        job.completed_at = datetime.utcnow()
+        return job
+    
+    async def _process_animation_prompt(self, job: GeminiJob, service: GeminiService, input_data: dict) -> GeminiJob:
+        """Process animation prompt generation (synchronous)."""
+        result = await service.generate_animation_prompt(
+            base64_image=input_data.get("base64_image", ""),
+            mime_type=input_data.get("mime_type", "image/jpeg"),
+            custom_prompt=input_data.get("custom_prompt")
+        )
+        job.status = "completed"
+        job.output_data = {"prompt": result}
+        job.completed_at = datetime.utcnow()
+        return job
+
+
+# Singleton pool instance
+_pool: Optional[PriorityWorkerPool] = None
+
+
+def get_pool() -> PriorityWorkerPool:
+    """Get the global Gemini worker pool instance."""
+    global _pool
+    if _pool is None:
+        _pool = PriorityWorkerPool(
+            database_url=DATABASE_URL,
+            job_model=GeminiJob,
+            job_processor=GeminiJobProcessor(),
+            config=WorkerConfig.from_env()
+        )
+    return _pool
+
+
+async def start_worker():
+    """Start the Gemini job worker pool."""
+    pool = get_pool()
+    await pool.start()
+
+
+async def stop_worker():
+    """Stop the Gemini job worker pool."""
+    pool = get_pool()
+    await pool.stop()

services/{job_worker.py → priority_worker_pool.py}

@@ -1,67 +1,163 @@
 """
-Priority-Tier Worker Pool for processing Gemini jobs.
+Modular Priority-Tier Worker Pool
 
-Architecture:
-- 15 workers total: 5 fast (5s), 5 medium (30s), 5 slow (60s)
-- Workers pick jobs based on priority tier
-- Jobs are rescheduled with next_process_at if third-party not done
-- No blocking on third-party polling
+A self-contained, plug-and-play worker pool for processing async jobs
+with priority-tier scheduling. Can be used in any Python application.
+
+Usage:
+    from services.priority_worker_pool import PriorityWorkerPool, WorkerConfig
+    
+    # Define your job processor function
+    async def process_my_job(job, session):
+        # Process job and return updated job
+        job.status = "completed"
+        job.output_data = {"result": "done"}
+        return job
+    
+    # Configure and start pool
+    pool = PriorityWorkerPool(
+        database_url="sqlite+aiosqlite:///./my_db.db",
+        job_model=MyJobModel,
+        job_processor=process_my_job,
+        config=WorkerConfig(fast_workers=5, medium_workers=5, slow_workers=5)
+    )
+    await pool.start()
+
+Environment Variables (optional):
+    FAST_WORKERS: Number of fast workers (default: 5)
+    MEDIUM_WORKERS: Number of medium workers (default: 5)
+    SLOW_WORKERS: Number of slow workers (default: 5)
+    FAST_INTERVAL: Fast tier polling interval in seconds (default: 5)
+    MEDIUM_INTERVAL: Medium tier polling interval in seconds (default: 30)
+    SLOW_INTERVAL: Slow tier polling interval in seconds (default: 60)
+
+Dependencies:
+    sqlalchemy[asyncio]>=2.0.0
+    aiosqlite (for SQLite) or asyncpg (for PostgreSQL)
+
+Job Model Requirements:
+    Your job model must have these columns:
+    - job_id: str (unique identifier)
+    - status: str (queued, processing, completed, failed, cancelled)
+    - priority: str (fast, medium, slow)
+    - next_process_at: datetime (nullable, for rescheduling)
+    - retry_count: int (default 0)
+    - created_at: datetime
+    - started_at: datetime (nullable)
+    - completed_at: datetime (nullable)
+    - error_message: str (nullable)
 """
 import asyncio
 import logging
 import os
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
 from datetime import datetime, timedelta
-from typing import Optional, List
+from typing import Optional, List, Callable, Any, TypeVar, Generic
 from sqlalchemy import select, or_, and_
 from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
 
-from core.database import DATABASE_URL
-from core.models import GeminiJob
-from services.gemini_service import GeminiService
-
 logger = logging.getLogger(__name__)
 
-# Worker configuration
-FAST_WORKERS = int(os.getenv("FAST_WORKERS", "5"))
-MEDIUM_WORKERS = int(os.getenv("MEDIUM_WORKERS", "5"))
-SLOW_WORKERS = int(os.getenv("SLOW_WORKERS", "5"))
+# Generic type for job model
+JobType = TypeVar('JobType')
 
-FAST_INTERVAL = int(os.getenv("FAST_INTERVAL", "5"))      # 5 seconds
-MEDIUM_INTERVAL = int(os.getenv("MEDIUM_INTERVAL", "30"))  # 30 seconds
-SLOW_INTERVAL = int(os.getenv("SLOW_INTERVAL", "60"))      # 60 seconds
 
-# Job type to priority mapping
-JOB_PRIORITY_MAP = {
-    "text": "fast",
-    "analyze": "fast",
-    "animation_prompt": "fast",
-    "image": "medium",
-    "edit_image": "medium",
-    "video": "slow"
-}
+@dataclass
+class WorkerConfig:
+    """Configuration for the worker pool."""
+    fast_workers: int = 5
+    medium_workers: int = 5
+    slow_workers: int = 5
+    fast_interval: int = 5      # seconds
+    medium_interval: int = 30   # seconds
+    slow_interval: int = 60     # seconds
+    max_retries: int = 60       # Max retry attempts before failing
+    
+    @classmethod
+    def from_env(cls) -> 'WorkerConfig':
+        """Create config from environment variables."""
+        return cls(
+            fast_workers=int(os.getenv("FAST_WORKERS", "5")),
+            medium_workers=int(os.getenv("MEDIUM_WORKERS", "5")),
+            slow_workers=int(os.getenv("SLOW_WORKERS", "5")),
+            fast_interval=int(os.getenv("FAST_INTERVAL", "5")),
+            medium_interval=int(os.getenv("MEDIUM_INTERVAL", "30")),
+            slow_interval=int(os.getenv("SLOW_INTERVAL", "60")),
+        )
 
-def get_priority_for_job_type(job_type: str) -> str:
-    """Get the priority tier for a job type."""
-    return JOB_PRIORITY_MAP.get(job_type, "fast")
 
-def get_interval_for_priority(priority: str) -> int:
-    """Get the polling interval in seconds for a priority tier."""
-    if priority == "fast":
-        return FAST_INTERVAL
-    elif priority == "medium":
-        return MEDIUM_INTERVAL
-    else:
-        return SLOW_INTERVAL
+@dataclass
+class PriorityMapping:
+    """Maps job types to priority tiers."""
+    mappings: dict = field(default_factory=dict)
+    
+    def get_priority(self, job_type: str, default: str = "fast") -> str:
+        """Get priority for a job type."""
+        return self.mappings.get(job_type, default)
+    
+    def get_interval(self, priority: str, config: WorkerConfig) -> int:
+        """Get polling interval for a priority tier."""
+        if priority == "fast":
+            return config.fast_interval
+        elif priority == "medium":
+            return config.medium_interval
+        else:
+            return config.slow_interval
 
 
-class PriorityWorker:
+class JobProcessor(ABC, Generic[JobType]):
+    """Abstract base class for job processors."""
+    
+    @abstractmethod
+    async def process(self, job: JobType, session: AsyncSession) -> JobType:
+        """
+        Process a job and return the updated job.
+        
+        Args:
+            job: The job to process
+            session: Database session for updates
+            
+        Returns:
+            The updated job with new status/output
+        """
+        pass
+    
+    @abstractmethod
+    async def check_status(self, job: JobType, session: AsyncSession) -> JobType:
+        """
+        Check status of an in-progress job (for async third-party operations).
+        
+        Args:
+            job: The job to check
+            session: Database session for updates
+            
+        Returns:
+            The updated job. Set next_process_at to reschedule if not done.
+        """
+        pass
+
+
+class PriorityWorker(Generic[JobType]):
     """Worker that processes jobs of a specific priority tier."""
     
-    def __init__(self, worker_id: int, priority: str, poll_interval: int, session_maker):
+    def __init__(
+        self,
+        worker_id: int,
+        priority: str,
+        poll_interval: int,
+        session_maker: async_sessionmaker,
+        job_model: type,
+        job_processor: JobProcessor[JobType],
+        max_retries: int = 60
+    ):
         self.worker_id = worker_id
         self.priority = priority
         self.poll_interval = poll_interval
         self.session_maker = session_maker
+        self.job_model = job_model
+        self.job_processor = job_processor
+        self.max_retries = max_retries
         self._running = False
         self._current_job_id: Optional[str] = None
     
@@ -88,24 +184,25 @@ class PriorityWorker:
     async def _process_one_job(self):
         """Find and process one job."""
         async with self.session_maker() as session:
-            # Find a job to process
             now = datetime.utcnow()
-            query = select(GeminiJob).where(
+            
+            # Query for jobs matching this priority tier
+            query = select(self.job_model).where(
                 and_(
-                    GeminiJob.priority == self.priority,
-                    GeminiJob.status.in_(["queued", "processing"]),
+                    self.job_model.priority == self.priority,
+                    self.job_model.status.in_(["queued", "processing"]),
                     or_(
-                        GeminiJob.next_process_at.is_(None),
-                        GeminiJob.next_process_at <= now
+                        self.job_model.next_process_at.is_(None),
+                        self.job_model.next_process_at <= now
                     )
                 )
-            ).order_by(GeminiJob.created_at).limit(1)
+            ).order_by(self.job_model.created_at).limit(1)
             
             result = await session.execute(query)
             job = result.scalar_one_or_none()
             
             if not job:
-                return  # No jobs to process
+                return
             
             self._current_job_id = job.job_id
             
@@ -120,142 +217,60 @@ class PriorityWorker:
             finally:
                 self._current_job_id = None
     
-    async def _process_job(self, session: AsyncSession, job: GeminiJob):
+    async def _process_job(self, session: AsyncSession, job: JobType):
         """Process a single job."""
-        logger.info(f"Worker {self.worker_id}: Processing job {job.job_id} (type: {job.job_type}, status: {job.status})")
-        
-        service = GeminiService()
-        input_data = job.input_data or {}
+        logger.info(f"Worker {self.worker_id}: Processing job {job.job_id} (status: {job.status})")
         
-        # If queued, start the operation
         if job.status == "queued":
+            # New job - start processing
             job.status = "processing"
             job.started_at = datetime.utcnow()
             await session.commit()
             
-            # Start the third-party operation
-            await self._start_operation(session, job, service, input_data)
-        
-        # Check status for operations that need polling (video)
-        if job.job_type == "video" and job.third_party_id:
-            await self._check_video_status(session, job, service)
-        # For synchronous operations (already completed in _start_operation)
-        # Nothing more to do
-    
-    async def _start_operation(self, session: AsyncSession, job: GeminiJob, service: GeminiService, input_data: dict):
-        """Start the third-party operation based on job type."""
-        if job.job_type == "video":
-            # Start async video generation
-            result = await service.start_video_generation(
-                base64_image=input_data.get("base64_image", ""),
-                mime_type=input_data.get("mime_type", "image/jpeg"),
-                prompt=input_data.get("prompt", ""),
-                aspect_ratio=input_data.get("aspect_ratio", "16:9"),
-                resolution=input_data.get("resolution", "720p"),
-                number_of_videos=input_data.get("number_of_videos", 1)
-            )
-            job.third_party_id = result.get("gemini_operation_name")
-            # Schedule first status check
-            job.next_process_at = datetime.utcnow() + timedelta(seconds=self.poll_interval)
-            await session.commit()
-            
-        elif job.job_type == "image":
-            # Synchronous image edit
-            result = await service.edit_image(
-                base64_image=input_data.get("base64_image", ""),
-                mime_type=input_data.get("mime_type", "image/jpeg"),
-                prompt=input_data.get("prompt", "")
-            )
-            job.status = "completed"
-            job.output_data = {"image": result}
-            job.completed_at = datetime.utcnow()
-            await session.commit()
-            
-        elif job.job_type == "text":
-            # Synchronous text generation
-            result = await service.generate_text(
-                prompt=input_data.get("prompt", ""),
-                model=input_data.get("model")
-            )
-            job.status = "completed"
-            job.output_data = {"text": result}
-            job.completed_at = datetime.utcnow()
-            await session.commit()
-            
-        elif job.job_type == "analyze":
-            # Synchronous image analysis
-            result = await service.analyze_image(
-                base64_image=input_data.get("base64_image", ""),
-                mime_type=input_data.get("mime_type", "image/jpeg"),
-                prompt=input_data.get("prompt", "")
-            )
-            job.status = "completed"
-            job.output_data = {"analysis": result}
-            job.completed_at = datetime.utcnow()
-            await session.commit()
-            
-        elif job.job_type == "animation_prompt":
-            # Synchronous animation prompt generation
-            result = await service.generate_animation_prompt(
-                base64_image=input_data.get("base64_image", ""),
-                mime_type=input_data.get("mime_type", "image/jpeg"),
-                custom_prompt=input_data.get("custom_prompt")
-            )
-            job.status = "completed"
-            job.output_data = {"prompt": result}
-            job.completed_at = datetime.utcnow()
-            await session.commit()
+            # Process the job
+            job = await self.job_processor.process(job, session)
         else:
+            # Already processing - check status
+            job = await self.job_processor.check_status(job, session)
+        
+        # Handle retry limit
+        if job.status == "processing" and job.retry_count > self.max_retries:
             job.status = "failed"
-            job.error_message = f"Unknown job type: {job.job_type}"
+            job.error_message = f"Max retries ({self.max_retries}) exceeded"
             job.completed_at = datetime.utcnow()
-            await session.commit()
-    
-    async def _check_video_status(self, session: AsyncSession, job: GeminiJob, service: GeminiService):
-        """Check video generation status and reschedule if not done."""
-        try:
-            status_result = await service.check_video_status(job.third_party_id)
-            
-            if status_result.get("done"):
-                if status_result.get("status") == "completed":
-                    # Download video
-                    video_url = status_result.get("video_url")
-                    if video_url:
-                        filename = await service.download_video(video_url, job.job_id)
-                        job.status = "completed"
-                        job.output_data = {"filename": filename}
-                    else:
-                        job.status = "failed"
-                        job.error_message = "No video URL returned"
-                else:
-                    job.status = "failed"
-                    job.error_message = status_result.get("error", "Unknown error")
-                
-                job.completed_at = datetime.utcnow()
-            else:
-                # Not done - reschedule
-                job.retry_count += 1
-                job.next_process_at = datetime.utcnow() + timedelta(seconds=self.poll_interval)
-                logger.debug(f"Job {job.job_id}: Not done, retry #{job.retry_count}, next check at {job.next_process_at}")
-            
-            await session.commit()
-            
-        except Exception as e:
-            logger.error(f"Error checking video status for {job.job_id}: {e}")
-            job.retry_count += 1
-            job.next_process_at = datetime.utcnow() + timedelta(seconds=self.poll_interval)
-            if job.retry_count > 60:  # ~1 hour of retries
-                job.status = "failed"
-                job.error_message = f"Max retries exceeded: {str(e)}"
-                job.completed_at = datetime.utcnow()
-            await session.commit()
+        
+        await session.commit()
 
 
-class WorkerPool:
-    """Pool of priority-tier workers."""
+class PriorityWorkerPool(Generic[JobType]):
+    """
+    Modular priority-tier worker pool.
     
-    def __init__(self):
-        self.engine = create_async_engine(DATABASE_URL, echo=False)
+    Can be used with any job model that follows the required schema.
+    """
+    
+    def __init__(
+        self,
+        database_url: str,
+        job_model: type,
+        job_processor: JobProcessor[JobType],
+        config: Optional[WorkerConfig] = None
+    ):
+        """
+        Initialize the worker pool.
+        
+        Args:
+            database_url: SQLAlchemy async database URL
+            job_model: Your ORM model class for jobs
+            job_processor: Instance of JobProcessor to handle jobs
+            config: Worker configuration (uses env vars if not provided)
+        """
+        self.database_url = database_url
+        self.job_model = job_model
+        self.job_processor = job_processor
+        self.config = config or WorkerConfig.from_env()
+        
+        self.engine = create_async_engine(database_url, echo=False)
         self.session_maker = async_sessionmaker(
             self.engine,
             class_=AsyncSession,
@@ -270,55 +285,76 @@ class WorkerPool:
         worker_id = 0
         
         # Create fast workers
-        for i in range(FAST_WORKERS):
-            worker = PriorityWorker(worker_id, "fast", FAST_INTERVAL, self.session_maker)
+        for i in range(self.config.fast_workers):
+            worker = PriorityWorker(
+                worker_id=worker_id,
+                priority="fast",
+                poll_interval=self.config.fast_interval,
+                session_maker=self.session_maker,
+                job_model=self.job_model,
+                job_processor=self.job_processor,
+                max_retries=self.config.max_retries
+            )
             self.workers.append(worker)
             await worker.start()
             worker_id += 1
         
         # Create medium workers
-        for i in range(MEDIUM_WORKERS):
-            worker = PriorityWorker(worker_id, "medium", MEDIUM_INTERVAL, self.session_maker)
+        for i in range(self.config.medium_workers):
+            worker = PriorityWorker(
+                worker_id=worker_id,
+                priority="medium",
+                poll_interval=self.config.medium_interval,
+                session_maker=self.session_maker,
+                job_model=self.job_model,
+                job_processor=self.job_processor,
+                max_retries=self.config.max_retries
+            )
             self.workers.append(worker)
             await worker.start()
             worker_id += 1
         
         # Create slow workers
-        for i in range(SLOW_WORKERS):
-            worker = PriorityWorker(worker_id, "slow", SLOW_INTERVAL, self.session_maker)
+        for i in range(self.config.slow_workers):
+            worker = PriorityWorker(
+                worker_id=worker_id,
+                priority="slow",
+                poll_interval=self.config.slow_interval,
+                session_maker=self.session_maker,
+                job_model=self.job_model,
+                job_processor=self.job_processor,
+                max_retries=self.config.max_retries
+            )
             self.workers.append(worker)
             await worker.start()
             worker_id += 1
         
-        logger.info(f"WorkerPool started with {len(self.workers)} workers: {FAST_WORKERS} fast, {MEDIUM_WORKERS} medium, {SLOW_WORKERS} slow")
+        total = self.config.fast_workers + self.config.medium_workers + self.config.slow_workers
+        logger.info(
+            f"PriorityWorkerPool started with {total} workers: "
+            f"{self.config.fast_workers} fast, {self.config.medium_workers} medium, {self.config.slow_workers} slow"
+        )
     
     async def stop(self):
         """Stop all workers."""
         self._running = False
         for worker in self.workers:
             await worker.stop()
-        logger.info("WorkerPool stopped")
-
-
-# Singleton pool instance
-_pool: Optional[WorkerPool] = None
+        logger.info("PriorityWorkerPool stopped")
 
 
-def get_pool() -> WorkerPool:
-    """Get the global worker pool instance."""
-    global _pool
-    if _pool is None:
-        _pool = WorkerPool()
-    return _pool
+# Convenience functions for priority mapping
+def get_priority_for_job_type(job_type: str, mappings: dict) -> str:
+    """Get priority tier for a job type using provided mappings."""
+    return mappings.get(job_type, "fast")
 
 
-async def start_worker():
-    """Start the background worker pool."""
-    pool = get_pool()
-    await pool.start()
-
-
-async def stop_worker():
-    """Stop the background worker pool."""
-    pool = get_pool()
-    await pool.stop()
+def get_interval_for_priority(priority: str, config: Optional[WorkerConfig] = None) -> int:
+    """Get polling interval for a priority tier."""
+    cfg = config or WorkerConfig.from_env()
+    if priority == "fast":
+        return cfg.fast_interval
+    elif priority == "medium":
+        return cfg.medium_interval
+    else:
+        return cfg.slow_interval

tests/test_worker_pool.py

@@ -7,13 +7,19 @@ import asyncio
 from unittest.mock import patch, MagicMock, AsyncMock
 from datetime import datetime, timedelta
 
-# Test the priority mapping
-from services.job_worker import (
-    get_priority_for_job_type,
-    get_interval_for_priority,
+# Test the modular priority worker pool
+from services.priority_worker_pool import (
+    PriorityWorkerPool,
     PriorityWorker,
-    WorkerPool,
-    JOB_PRIORITY_MAP
+    WorkerConfig,
+    get_interval_for_priority
+)
+
+# Test the Gemini-specific implementation
+from services.gemini_job_worker import (
+    get_priority_for_job_type,
+    JOB_PRIORITY_MAP,
+    GeminiJobProcessor
 )
 
 
@@ -68,48 +74,37 @@ class TestJobPriorityMap:
 
 
 class TestWorkerPoolConfiguration:
-    """Test worker pool creates correct number of workers."""
-    
-    @pytest.mark.asyncio
-    async def test_creates_15_workers(self):
-        """Test that WorkerPool creates 15 workers (5 fast, 5 medium, 5 slow)."""
-        pool = WorkerPool()
-        
-        # Start pool (workers will be created)
-        # Mock to prevent actual polling
-        with patch.object(PriorityWorker, '_poll_loop', new_callable=AsyncMock):
-            await pool.start()
-        
-        assert len(pool.workers) == 15
-        
-        # Count by priority
-        fast_workers = [w for w in pool.workers if w.priority == "fast"]
-        medium_workers = [w for w in pool.workers if w.priority == "medium"]
-        slow_workers = [w for w in pool.workers if w.priority == "slow"]
-        
-        assert len(fast_workers) == 5
-        assert len(medium_workers) == 5
-        assert len(slow_workers) == 5
-        
-        await pool.stop()
-    
-    @pytest.mark.asyncio
-    async def test_workers_have_correct_intervals(self):
-        """Test that workers have correct poll intervals."""
-        pool = WorkerPool()
-        
-        with patch.object(PriorityWorker, '_poll_loop', new_callable=AsyncMock):
-            await pool.start()
-        
-        for worker in pool.workers:
-            if worker.priority == "fast":
-                assert worker.poll_interval == 5
-            elif worker.priority == "medium":
-                assert worker.poll_interval == 30
-            elif worker.priority == "slow":
-                assert worker.poll_interval == 60
-        
-        await pool.stop()
+    """Test worker pool configuration."""
+    
+    def test_default_config(self):
+        """Test WorkerConfig defaults."""
+        config = WorkerConfig()
+        assert config.fast_workers == 5
+        assert config.medium_workers == 5
+        assert config.slow_workers == 5
+        assert config.fast_interval == 5
+        assert config.medium_interval == 30
+        assert config.slow_interval == 60
+    
+    def test_custom_config(self):
+        """Test WorkerConfig with custom values."""
+        config = WorkerConfig(
+            fast_workers=3,
+            medium_workers=2,
+            slow_workers=1,
+            fast_interval=10,
+            medium_interval=60,
+            slow_interval=120
+        )
+        assert config.fast_workers == 3
+        assert config.medium_workers == 2
+        assert config.slow_workers == 1
+    
+    def test_total_workers_calculation(self):
+        """Test total workers from config."""
+        config = WorkerConfig(fast_workers=5, medium_workers=5, slow_workers=5)
+        total = config.fast_workers + config.medium_workers + config.slow_workers
+        assert total == 15
 
 
 class TestPriorityWorker:
@@ -117,7 +112,15 @@ class TestPriorityWorker:
     
     def test_worker_has_correct_attributes(self):
         """Test worker initialization."""
-        worker = PriorityWorker(0, "fast", 5, None)
+        # PriorityWorker now requires more args, test with mocks
+        worker = PriorityWorker(
+            worker_id=0,
+            priority="fast",
+            poll_interval=5,
+            session_maker=None,
+            job_model=None,
+            job_processor=None
+        )
         
         assert worker.worker_id == 0
         assert worker.priority == "fast"