from celery import Celery from kombu import Exchange, Queue from app.core.config import get_settings settings = get_settings() # Two named queues: # high — small repos (< 10 MB), expected to complete in < 30s # low — large repos (>= 10 MB), may take several minutes # # Both are durable (survives Redis restart). The worker consumes high # first due to queue ordering in the -Q argument (see docker-compose.yml). default_exchange = Exchange("default", type="direct") HIGH_QUEUE = "high" LOW_QUEUE = "low" celery_app = Celery( "ai_code_review", broker=settings.redis_url, backend=settings.celery_result_backend or settings.redis_url, ) celery_app.conf.update( # Queue definitions task_queues=( Queue(HIGH_QUEUE, default_exchange, routing_key="high", durable=True), Queue(LOW_QUEUE, default_exchange, routing_key="low", durable=True), ), task_default_queue=HIGH_QUEUE, task_default_exchange="default", task_default_routing_key="high", # Route analyze_repository_task to high or low based on a size hint # passed as a task header (set in tasks.py before .apply_async()) task_routes={ "analyze_repository_task": {"queue": HIGH_QUEUE}, # default; overridden at call site }, # Serialisation task_serializer="json", result_serializer="json", accept_content=["json"], # Results expire after 24 h — matches the cache TTL result_expires=86400, # Eagerness (test mode) task_always_eager=settings.celery_task_always_eager, # Acknowledge only after the task completes, not on receipt. # Prevents a crashed worker from silently dropping a job. task_acks_late=True, task_reject_on_worker_lost=True, # Worker settings worker_prefetch_multiplier=1, # one task at a time per worker slot )