feat(api): migrate from Celery to RQ and expose queue position

Replace Celery with RQ to allow reporting pending task position and
in-progress count in task status responses, resolving #107.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: vvatelot

bases/ecoindex/backend/routers/tasks.py

@@ -1,10 +1,8 @@
-from json import loads
 from typing import Annotated
 from urllib.parse import urlparse, urlunparse
 
 import idna
 import requests
-from celery.result import AsyncResult
 from ecoindex.backend.dependencies.validation import validate_api_key_batch
 from ecoindex.backend.models.dependencies_parameters.id import IdParameter
 from ecoindex.backend.utils import check_quota
@@ -17,10 +15,22 @@
     example_daily_limit_response,
     example_host_unreachable,
 )
-from ecoindex.models.tasks import QueueTaskApi, QueueTaskApiBatch, QueueTaskResult
+from ecoindex.models.tasks import QueueTaskApi, QueueTaskApiBatch
 from ecoindex.scraper.scrap import EcoindexScraper
 from ecoindex.worker.tasks import ecoindex_batch_import_task, ecoindex_task
-from ecoindex.worker_component import app as task_app
+from ecoindex.worker_component import (
+    NoSuchJobError,
+    cancel_job,
+    ecoindex_batch_queue,
+    ecoindex_queue,
+    fetch_job,
+    get_queue_for_job,
+    get_queue_position,
+    get_retry_policy,
+    get_tasks_in_progress,
+    map_job_status_to_task_status,
+    parse_job_result,
+)
 from fastapi import APIRouter, Depends, HTTPException, Response, status
 from fastapi.params import Body
 from sqlmodel.ext.asyncio.session import AsyncSession
@@ -70,6 +80,18 @@ def convert_url_to_punycode(url: str) -> str:
         return url
 
 
+def _enqueue_settings(*, with_retry: bool = True) -> dict:
+    settings = Settings()
+    enqueue_settings = {
+        "result_ttl": settings.RQ_RESULT_TTL,
+        "failure_ttl": settings.RQ_FAILURE_TTL,
+        "job_timeout": settings.RQ_JOB_TIMEOUT,
+    }
+    if with_retry:
+        enqueue_settings["retry"] = get_retry_policy()
+    return enqueue_settings
+
+
 @router.post(
     name="Add new ecoindex analysis task to the waiting queue",
     path="/",
@@ -142,14 +164,16 @@ async def add_ecoindex_analysis_task(
             detail=f"The URL {web_page.url} is unreachable. Are you really sure of this url? 🤔 ({e.response.status_code if e.response else ''})",
         )
 
-    task_result = ecoindex_task.delay(  # type: ignore
+    job = ecoindex_queue.enqueue(
+        ecoindex_task,
         url=str(web_page.url),
         width=web_page.width,
         height=web_page.height,
         custom_headers=headers,
+        **_enqueue_settings(),
     )
 
-    return task_result.id
+    return job.id
 
 
 @router.get(
@@ -166,23 +190,35 @@ async def get_ecoindex_analysis_task_by_id(
     response: Response,
     id: IdParameter,
 ) -> QueueTaskApi:
-    t = AsyncResult(id=str(id), app=task_app)
+    try:
+        job = fetch_job(str(id))
+    except NoSuchJobError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Task not found",
+        ) from exc
+
+    queue = get_queue_for_job(job)
+    task_status = map_job_status_to_task_status(job)
+    result = parse_job_result(job)
 
     task_response = QueueTaskApi(
-        id=str(t.id),
-        status=t.state,
+        id=job.id,
+        status=task_status,
+        queue_position=get_queue_position(job, queue),
+        tasks_in_progress=get_tasks_in_progress(queue),
     )
 
-    if t.state == TaskStatus.PENDING:
+    if task_status == TaskStatus.PENDING:
         response.status_code = status.HTTP_425_TOO_EARLY
 
         return task_response
 
-    if t.state == TaskStatus.SUCCESS:
-        task_response.ecoindex_result = QueueTaskResult(**loads(t.result))
+    if task_status == TaskStatus.SUCCESS:
+        task_response.ecoindex_result = result
 
-    if t.state == TaskStatus.FAILURE:
-        task_response.task_error = t.info
+    if task_status == TaskStatus.FAILURE:
+        task_response.task_error = result
 
     response.status_code = status.HTTP_200_OK
 
@@ -198,9 +234,13 @@ async def get_ecoindex_analysis_task_by_id(
 async def delete_ecoindex_analysis_task_by_id(
     id: IdParameter,
 ) -> None:
-    res = task_app.control.revoke(id, terminate=True, signal="SIGKILL")
-
-    return res
+    try:
+        cancel_job(str(id))
+    except NoSuchJobError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Task not found",
+        ) from exc
 
 
 @router.post(
@@ -227,12 +267,14 @@ async def add_ecoindex_analysis_task_batch(
     ],
     batch_key: str = Depends(validate_api_key_batch),
 ):
-    task_result = ecoindex_batch_import_task.delay(  # type: ignore
+    job = ecoindex_batch_queue.enqueue(
+        ecoindex_batch_import_task,
         results=[result.model_dump() for result in results],
         source=batch_key["source"],  # type: ignore
+        **_enqueue_settings(with_retry=False),
     )
 
-    return task_result.id
+    return job.id
 
 
 @router.get(
@@ -250,14 +292,22 @@ async def get_ecoindex_analysis_batch_task_by_id(
     id: IdParameter,
     _: str = Depends(validate_api_key_batch),
 ) -> QueueTaskApiBatch:
-    t = AsyncResult(id=str(id), app=task_app)
+    try:
+        job = fetch_job(str(id))
+    except NoSuchJobError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Task not found",
+        ) from exc
+
+    task_status = map_job_status_to_task_status(job)
 
     task_response = QueueTaskApiBatch(
-        id=str(t.id),
-        status=t.state,
+        id=job.id,
+        status=task_status,
     )
 
-    if t.state == TaskStatus.PENDING:
+    if task_status == TaskStatus.PENDING:
         response.status_code = status.HTTP_425_TOO_EARLY
 
     return task_response

bases/ecoindex/worker/health.py

@@ -1,18 +1,23 @@
 from ecoindex.models.api import HealthWorker, HealthWorkers
-from ecoindex.worker.tasks import app
+from ecoindex.worker_component import redis_connection
+from redis.exceptions import ConnectionError as RedisConnectionError
+from rq.worker import Worker
 
 
 def is_worker_healthy() -> HealthWorkers:
-    workers = []
-    workers_ping = app.control.ping()
-
-    for worker in workers_ping:
-        for name in worker:
+    try:
+        workers = []
+        for worker in Worker.all(connection=redis_connection):
             workers.append(
-                HealthWorker(name=name, healthy=True if "ok" in worker[name] else False)
+                HealthWorker(
+                    name=worker.name,
+                    healthy=worker.state in ("busy", "idle"),
+                )
             )
+    except RedisConnectionError:
+        return HealthWorkers(healthy=False, workers=[])
 
     return HealthWorkers(
-        healthy=False if False in [w.healthy for w in workers] or not workers else True,
+        healthy=bool(workers) and all(w.healthy for w in workers),
         workers=workers,
     )

bases/ecoindex/worker/tasks.py

@@ -1,6 +1,7 @@
 from asyncio import run
 from os import getcwd
 from urllib.parse import urlparse
+from uuid import UUID
 
 from ecoindex.backend.utils import check_quota, format_exception_response
 from ecoindex.config.settings import Settings
@@ -19,38 +20,39 @@
 from ecoindex.models.enums import TaskStatus
 from ecoindex.models.tasks import QueueTaskError, QueueTaskResult
 from ecoindex.scraper.scrap import EcoindexScraper
-from ecoindex.worker_component import app
 from playwright._impl._errors import Error as WebDriverException
+from rq import get_current_job
 from sentry_sdk import init as sentry_init
 
 if Settings().GLITCHTIP_DSN:
     sentry_init(Settings().GLITCHTIP_DSN)
 
 
-@app.task(
-    name="ecoindex.analysis",
-    bind=True,
-    autoretry_for=(Exception,),
-    retry_backoff=5,
-    retry_kwargs={"max_retries": 5},
-    timezone=Settings().TZ,
-    queue="ecoindex",
-    dont_autoretry_for=[EcoindexScraperStatusException, TypeError],
-)
+def _get_task_id() -> UUID:
+    job = get_current_job()
+    if job is None:
+        raise RuntimeError("No RQ job context available")
+    return UUID(job.id)
+
+
 def ecoindex_task(
-    self, url: str, width: int, height: int, custom_headers: dict[str, str]
+    url: str, width: int, height: int, custom_headers: dict[str, str]
 ) -> str:
     queue_task_result = run(
         async_ecoindex_task(
-            self, url=url, width=width, height=height, custom_headers=custom_headers
+            task_id=_get_task_id(),
+            url=url,
+            width=width,
+            height=height,
+            custom_headers=custom_headers,
         )
     )
 
     return queue_task_result.model_dump_json()
 
 
 async def async_ecoindex_task(
-    self,
+    task_id: UUID,
     url: str,
     width: int,
     height: int,
@@ -68,7 +70,7 @@ async def async_ecoindex_task(
             wait_after_scroll=Settings().WAIT_AFTER_SCROLL,
             wait_before_scroll=Settings().WAIT_BEFORE_SCROLL,
             screenshot=ScreenShot(
-                id=str(self.request.id), folder=f"{getcwd()}/screenshots/v1"
+                id=str(task_id), folder=f"{getcwd()}/screenshots/v1"
             )
             if Settings().ENABLE_SCREENSHOT
             else None,
@@ -79,7 +81,7 @@ async def async_ecoindex_task(
 
         db_result = await save_ecoindex_result_db(
             session=session,
-            id=self.request.id,
+            id=task_id,
             ecoindex_result=ecoindex,
         )
 
@@ -164,13 +166,7 @@ async def async_ecoindex_task(
         )
 
 
-@app.task(
-    name="ecoindex.batch_import",
-    timezone=Settings().TZ,
-    queue="ecoindex_batch",
-    bind=True,
-)
-def ecoindex_batch_import_task(self, results: list[dict], source: str):
+def ecoindex_batch_import_task(results: list[dict], source: str) -> str:
     queue_task_result = run(
         async_ecoindex_batch_import_task(
             results=[ApiEcoindex.model_validate(result) for result in results],

components/ecoindex/config/settings.py

@@ -20,6 +20,10 @@ class Settings(BaseSettings):
     FRONTEND_BASE_URL: str = "https://www.ecoindex.fr"
     GLITCHTIP_DSN: str = ""
     REDIS_CACHE_HOST: str = "localhost"
+    RQ_FAILURE_TTL: int = 86400
+    RQ_JOB_TIMEOUT: int = 600
+    RQ_RESULT_TTL: int = 86400
+    RQ_WORKERS: int = 3
     SCREENSHOTS_GID: int | None = None
     SCREENSHOTS_UID: int | None = None
     TZ: str = "Europe/Paris"

components/ecoindex/models/tasks.py

@@ -46,6 +46,15 @@ class QueueTaskApi(BaseModel):
         default=...,
         title="Status of the current task. Can be PENDING, FAILURE, SUCCESS",
     )
+    queue_position: int | None = Field(
+        default=None,
+        title="Position of the task in the waiting queue (0 = next to run)",
+        description="Null when the task is running or already completed.",
+    )
+    tasks_in_progress: int = Field(
+        default=0,
+        title="Number of tasks currently being processed on this queue",
+    )
     ecoindex_result: QueueTaskResult | None = Field(
         default=None, title="Result of the Ecoindex analysis"
     )