Spaces:
Running
Running
File size: 1,578 Bytes
e8a6607 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 | # import asyncio
# from typing import Dict, Any
# from bullmq import Queue
# import logging
# logger = logging.getLogger(__name__)
# class QueueManager:
# def __init__(self, queue_name: str, redis_url: str = "redis://127.0.0.1:6379"):
# self.queue_name = queue_name
# self.redis_url = redis_url
# self.queue = None
# async def connect(self) -> None:
# """Initializes the connection to the BullMQ Redis queue."""
# if not self.queue:
# # Passing connection string via opts dict
# self.queue = Queue(self.queue_name, opts={"connection": self.redis_url})
# logger.info(f"📡 Connected to Queue: {self.queue_name}")
# async def add_job(self, job_name: str, data: Dict[str, Any], priority: int = 0) -> Any:
# """Adds a job into the queue with optional configurations."""
# if not self.queue:
# await self.connect()
# # Inject the model string cleanly into the payload dictionary
# opts = {"priority": priority} if priority > 0 else {}
# # Call queue.add using only the valid signature parameters
# job = await self.queue.add(job_name, data, opts)
# logger.debug(f"➕ Job added! ID: {job.id} | Name: {job_name}")
# return job
# async def disconnect(self) -> None:
# """Gracefully closes the queue connections."""
# if self.queue:
# await self.queue.close()
# self.queue = None
# logger.info(f"🛑 Disconnected from Queue: {self.queue_name}")
|