File size: 2,676 Bytes
8e73bed 90ad38a 8e73bed 2fe5726 8e73bed 90ad38a 8e73bed bc06797 8e73bed 2fe5726 8e73bed |
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 |
"""
RQ Worker Entry Point
"""
import sys
import os
from pathlib import Path
from loguru import logger
from redis import Redis
from rq import Worker, Queue
from app.config import settings
from app.core.redis_client import get_redis_client
# Configure logging
logger.remove()
logger.add(
sys.stdout,
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan> - <level>{message}</level>",
level=settings.LOG_LEVEL
)
# Create logs directory if it doesn't exist
try:
log_dir = Path("logs")
log_dir.mkdir(parents=True, exist_ok=True)
logger.add(
"logs/swara_worker_{time:YYYY-MM-DD}.log",
rotation="1 day",
retention="7 days",
format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function} - {message}",
level=settings.LOG_LEVEL
)
except (PermissionError, OSError) as e:
# If we can't write to logs directory, only use stdout
logger.warning(f"Cannot create log file: {e}. Using stdout only.")
def main():
"""Start RQ worker"""
logger.info("=" * 70)
logger.info("🔧 SWARA Worker Starting...")
logger.info("=" * 70)
logger.info(f"Environment: {settings.ENV}")
logger.info(f"Redis URL: {settings.REDIS_URL}")
logger.info(f"Queue Name: {settings.TASK_QUEUE_NAME}")
try:
# Connect to Redis (decode_responses=False untuk RQ 2.0 compatibility)
# Gunakan connection baru, jangan pakai global redis_client
logger.info("Creating worker Redis connection...")
redis_conn = Redis.from_url(
settings.REDIS_URL,
decode_responses=False, # PENTING: RQ butuh bytes, bukan string!
max_connections=10,
socket_connect_timeout=30,
socket_timeout=30,
socket_keepalive=True,
health_check_interval=30
)
redis_conn.ping()
logger.info("✓ Redis connected")
# Create queue
queue = Queue(settings.TASK_QUEUE_NAME, connection=redis_conn)
logger.info(f"✓ Queue '{settings.TASK_QUEUE_NAME}' initialized")
logger.info("=" * 70)
logger.info("✓ Worker ready and listening for tasks...")
logger.info("=" * 70)
# Start worker (rq 2.0+ doesn't need Connection context)
worker = Worker([queue], connection=redis_conn)
worker.work(with_scheduler=True)
except KeyboardInterrupt:
logger.info("\n🛑 Worker interrupted by user")
except Exception as e:
logger.error(f"❌ Worker failed: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
|