diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py new file mode 100644 index 0000000..1d89c78 --- /dev/null +++ b/backend/app/celery_app.py @@ -0,0 +1,45 @@ +"""Celery configuration with separate queue lanes. + +Queues: + import — filesystem import path (FC-2) + ml — WD14 + SigLIP inference (FC-2; runs in the ml-worker image) + thumbnail — image and video thumbnail generation (FC-2) + download — gallery-dl tasks (FC-3) + scan — periodic source checks (FC-3) — kept separate so long imports + don't starve the scheduler + maintenance — pHash recomputation, centroid rebuild, etc. (FC-2/FC-3) + default — anything not explicitly routed +""" + +from celery import Celery + +from .config import get_config + + +def make_celery() -> Celery: + cfg = get_config() + app = Celery( + "fabledcurator", + broker=cfg.celery_broker_url, + backend=cfg.celery_result_backend, + include=["backend.app.tasks.smoke"], # FC-2/FC-3 extend this list + ) + app.conf.update( + task_default_queue="default", + task_routes={ + "backend.app.tasks.import_file.*": {"queue": "import"}, + "backend.app.tasks.ml.*": {"queue": "ml"}, + "backend.app.tasks.thumbnail.*": {"queue": "thumbnail"}, + "backend.app.tasks.download.*": {"queue": "download"}, + "backend.app.tasks.scan.*": {"queue": "scan"}, + "backend.app.tasks.maintenance.*": {"queue": "maintenance"}, + }, + # Heavy ML tasks need fair dispatch — see ImageRepo's precedent. + task_acks_late=True, + worker_prefetch_multiplier=1, + broker_connection_retry_on_startup=True, + ) + return app + + +celery = make_celery() diff --git a/backend/app/tasks/__init__.py b/backend/app/tasks/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/tasks/smoke.py b/backend/app/tasks/smoke.py new file mode 100644 index 0000000..ad452fb --- /dev/null +++ b/backend/app/tasks/smoke.py @@ -0,0 +1,8 @@ +"""Smoke task used during FC-1 to confirm Celery routing works end-to-end.""" + +from ..celery_app import celery + + +@celery.task(name="backend.app.tasks.smoke.ping") +def ping(value: str = "pong") -> str: + return value diff --git a/tests/test_celery_smoke.py b/tests/test_celery_smoke.py new file mode 100644 index 0000000..40b0054 --- /dev/null +++ b/tests/test_celery_smoke.py @@ -0,0 +1,24 @@ +"""Smoke test — celery runs the ping task in eager mode (no broker required).""" + +import pytest + +from backend.app.celery_app import celery +from backend.app.tasks.smoke import ping + + +@pytest.fixture(autouse=True) +def celery_eager(): + celery.conf.task_always_eager = True + celery.conf.task_eager_propagates = True + yield + celery.conf.task_always_eager = False + + +def test_ping_returns_pong(): + result = ping.delay() + assert result.get(timeout=1) == "pong" + + +def test_ping_with_value(): + result = ping.delay("hello") + assert result.get(timeout=1) == "hello"