Spaces:
Configuration error
Configuration error
File size: 7,310 Bytes
5292340 | 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 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 | """
Job Queue Module
Redis-backed job queue for async job processing.
Provides simple enqueue/dequeue operations with graceful degradation.
"""
import json
from typing import Optional, Any
from datetime import datetime, timezone
from app.core.config import get_settings
from app.core.redis import get_redis_client, RedisClient
from app.core.logging import get_logger
logger = get_logger(__name__)
settings = get_settings()
class JobQueueError(Exception):
"""Base exception for job queue errors."""
pass
class QueueUnavailableError(JobQueueError):
"""Raised when the queue is unavailable."""
pass
class JobQueue:
"""
Redis-backed job queue for distributed job processing.
Uses Redis LIST data structure for reliable queue semantics:
- RPUSH: Add job to queue (tail)
- BRPOP: Block and pop job from queue (tail -> FIFO with LPUSH)
- LPUSH + BRPOP: FIFO queue
Message format:
{
"job_id": int,
"enqueued_at": ISO timestamp,
"priority": int (optional, for future use)
}
"""
def __init__(
self,
queue_name: Optional[str] = None,
redis_client: Optional[RedisClient] = None
):
"""
Initialize job queue.
Args:
queue_name: Optional queue name (uses config default if not provided)
redis_client: Optional Redis client (uses singleton if not provided)
"""
self.queue_name = queue_name or settings.job_queue_name
self.redis = redis_client or get_redis_client()
# Processing queue for reliability (jobs being processed)
self.processing_queue = f"{self.queue_name}:processing"
@property
def is_available(self) -> bool:
"""Check if queue is available."""
return self.redis.is_available
def enqueue(self, job_id: int, priority: int = 0) -> bool:
"""
Add a job to the queue.
Args:
job_id: Job ID to enqueue
priority: Job priority (reserved for future use)
Returns:
True if successfully enqueued
Raises:
QueueUnavailableError: If Redis is unavailable
"""
if not self.is_available:
logger.error("Cannot enqueue job - queue unavailable")
raise QueueUnavailableError("Job queue is unavailable")
message = {
"job_id": job_id,
"enqueued_at": datetime.now(timezone.utc).isoformat(),
"priority": priority,
}
# LPUSH for FIFO with BRPOP
result = self.redis.lpush(self.queue_name, json.dumps(message))
if result is not None:
logger.info(f"Job enqueued: job_id={job_id}, queue_length={result}")
return True
logger.error(f"Failed to enqueue job: job_id={job_id}")
return False
def dequeue(self, timeout: int = 0) -> Optional[dict]:
"""
Remove and return a job from the queue.
Uses blocking pop for efficient polling.
Args:
timeout: Seconds to wait for job (0 = block forever)
Returns:
Job message dict or None if timeout/unavailable
"""
if not self.is_available:
logger.debug("Queue unavailable for dequeue")
return None
# BRPOP blocks until item available or timeout
result = self.redis.brpop([self.queue_name], timeout=timeout)
if result is None:
return None
try:
# result is (queue_name, value)
queue_name, message_str = result
message = json.loads(message_str)
logger.info(f"Job dequeued: job_id={message.get('job_id')}")
return message
except (json.JSONDecodeError, TypeError, ValueError) as e:
logger.error(f"Failed to parse queue message: {e}")
return None
def dequeue_nonblocking(self) -> Optional[dict]:
"""
Non-blocking dequeue.
Returns:
Job message dict or None if queue empty
"""
if not self.is_available:
return None
message_str = self.redis.rpop(self.queue_name)
if message_str is None:
return None
try:
message = json.loads(message_str)
logger.info(f"Job dequeued (non-blocking): job_id={message.get('job_id')}")
return message
except (json.JSONDecodeError, TypeError) as e:
logger.error(f"Failed to parse queue message: {e}")
return None
def peek(self) -> Optional[dict]:
"""
View next job without removing it.
Returns:
Job message dict or None if queue empty
"""
if not self.is_available:
return None
# LRANGE to get last element (BRPOP pops from right)
messages = self.redis.lrange(self.queue_name, -1, -1)
if not messages:
return None
try:
return json.loads(messages[0])
except (json.JSONDecodeError, TypeError):
return None
def size(self) -> int:
"""
Get current queue size.
Returns:
Number of jobs in queue
"""
return self.redis.llen(self.queue_name)
def clear(self) -> int:
"""
Clear all jobs from the queue.
Use with caution! Primarily for testing.
Returns:
Number of jobs cleared
"""
if not self.is_available:
return 0
count = self.size()
self.redis.delete(self.queue_name)
logger.warning(f"Queue cleared: {count} jobs removed")
return count
def requeue(self, job_id: int) -> bool:
"""
Re-add a failed job to the queue.
Used when job processing fails and needs retry.
Args:
job_id: Job ID to requeue
Returns:
True if successfully requeued
"""
return self.enqueue(job_id, priority=-1) # Lower priority for retries
def health_check(self) -> dict:
"""
Check queue health.
Returns:
Dictionary with status info
"""
available = self.is_available
return {
"available": available,
"queue_name": self.queue_name,
"size": self.size() if available else 0,
"redis": self.redis.health_check() if self.redis else {"status": "not_initialized"}
}
# =============================================================================
# Module-level singleton
# =============================================================================
_job_queue: Optional[JobQueue] = None
def get_job_queue() -> JobQueue:
"""
Get JobQueue singleton instance.
Returns:
JobQueue instance
"""
global _job_queue
if _job_queue is None:
_job_queue = JobQueue()
return _job_queue
|