✨(worker) switch from Celery to Dramatiq

Makefile

@@ -324,8 +324,9 @@ back-poetry-tree: ## show dependencies as a tree
 	@$(COMPOSE) run --rm --build backend-dev pipdeptree
 .PHONY: back-poetry-tree
 
-pip-audit: ## check the dependencies
-	@$(COMPOSE) run --rm --no-deps -e HOME=/tmp --build backend-dev pip-audit
+pip-audit: ## check the dependencies	
+	# https://github.com/pypa/pip/issues/13607
+	@$(COMPOSE) run --rm --no-deps -e HOME=/tmp --build backend-dev pip-audit --ignore-vuln GHSA-4xh5-x5gv-qwph
 .PHONY: pip-audit
 
 collectstatic: ## collect static files

Procfile

@@ -1,3 +1,4 @@
 web: bin/scalingo_run_web
-worker: celery -A messages.celery_app worker --task-events --beat -l INFO -c $CELERY_CONCURRENCY -Q celery,default
+worker: python manage.py worker
+scheduler: python manage.py crontab
 postdeploy: python manage.py migrate

compose.yaml

@@ -159,14 +159,14 @@ services:
       target: poetry
     pull_policy: build
 
-  celery-dev:
+  worker-dev:
     build:
       context: src/backend
       target: runtime-dev
       args:
         DOCKER_USER: ${DOCKER_USER:-1000}
     user: ${DOCKER_USER:-1000}
-    command: ["celery", "-A", "messages.celery_app", "worker", "-l", "DEBUG", "-Q", "celery,default"]
+    command: ["python", "manage.py", "worker", "-p", "2", "-t", "2"]
     environment:
       - DJANGO_CONFIGURATION=Development
     env_file:
@@ -177,39 +177,27 @@ services:
       - ./data/static:/data/static
     depends_on:
       - backend-dev
+      - redis
 
-  celery-ui:
+  scheduler-dev:
     build:
       context: src/backend
       target: runtime-dev
       args:
         DOCKER_USER: ${DOCKER_USER:-1000}
     user: ${DOCKER_USER:-1000}
-    depends_on:
-      - redis
+    command: ["python", "manage.py", "crontab"]
     environment:
-      - FLOWER_UNAUTHENTICATED_API=true
       - DJANGO_CONFIGURATION=Development
     env_file:
       - env.d/development/backend.defaults
       - env.d/development/backend.local
     volumes:
       - ./src/backend:/app
-    ports:
-      - "8903:8803"
-    command: celery -A messages.celery_app flower --port=8803
-
-  # nginx:
-  #   image: nginx:1.25
-  #   ports:
-  #     - "8083:8083"
-  #   volumes:
-  #     - ./docker/files/development/etc/nginx/conf.d:/etc/nginx/conf.d:ro
-  #   depends_on:
-  #     - keycloak
-  #     - backend-dev
-  #     - mta-in
-  #     - mta-out
+      - ./data/static:/data/static
+    depends_on:
+      - backend-dev
+      - redis
 
   frontend-dev:
     user: "${DOCKER_USER:-1000}"

docs/env.md

@@ -60,7 +60,7 @@ The application uses a new environment file structure with `.defaults` and `.loc
 | Variable | Default | Description | Required |
 |----------|---------|-------------|----------|
 | `REDIS_URL` | `redis://redis:6379` | Redis connection URL (internal) | Optional |
-| `CELERY_BROKER_URL` | `redis://redis:6379` | Celery message broker URL (internal) | Optional |
+| `DRAMATIQ_BROKER_URL` | `redis://redis:6379` | Dramatiq message broker URL (internal) | Optional |
 | `CACHES_DEFAULT_TIMEOUT` | `30` | Default cache timeout in seconds | Optional |
 
 **Note**: For external Redis access, use `localhost:8913`. For internal container communication, use `redis:6379`.

env.d/development/backend.defaults

@@ -27,6 +27,10 @@ METRICS_API_KEY=ExampleMetricsApiKey
 # Python
 PYTHONPATH=/app
 
+# Dramatiq
+DRAMATIQ_BROKER_URL=redis://redis:6379
+DRAMATIQ_CRONTAB_REDIS_URL=redis://redis:6379/0
+
 # Messages settings
 
 # Mail

src/backend/Dockerfile

@@ -1,8 +1,8 @@
 # https://hub.docker.com/_/python
-FROM python:3.13.7-slim-trixie AS base
+FROM python:3.13.9-slim-trixie AS base
 
 # Bump this to force an update of the apt repositories
-ENV MIN_UPDATE_DATE="2025-09-02"
+ENV MIN_UPDATE_DATE="2025-10-20"
 
 RUN apt-get update && apt-get upgrade -y \
     && rm -rf /var/lib/apt/lists/*
@@ -22,7 +22,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
 ENV POETRY_NO_INTERACTION=1
 ENV POETRY_VIRTUALENVS_CREATE=0
 ENV POETRY_VIRTUALENVS_OPTIONS_NO_PIP=1
-ENV POETRY_VERSION=2.1.4
+ENV POETRY_VERSION=2.2.1
 
 RUN python -m pip install poetry==${POETRY_VERSION}
 

src/backend/core/admin.py

@@ -1,15 +1,20 @@
 """Admin classes and registrations for core app."""
 
 from django.contrib import admin, messages
+from django.contrib.admin.views.decorators import staff_member_required
 from django.contrib.auth import admin as auth_admin
 from django.core.files.storage import storages
+from django.http import HttpResponse
 from django.shortcuts import redirect
 from django.template.response import TemplateResponse
-from django.urls import path
+from django.urls import path, reverse
 from django.utils.html import escape, format_html
 from django.utils.text import slugify
 from django.utils.translation import gettext_lazy as _
 
+import dramatiq
+import dramatiq_dashboard
+
 from core.api.utils import get_file_key
 from core.services.importer import ImportService
 
@@ -694,3 +699,75 @@ def get_html_body(self, obj):
         return obj.html_body
 
     get_html_body.short_description = "HTML Body"
+
+
+# Dramatiq Dashboard Integration
+@staff_member_required
+def dramatiq_dashboard_view(request):
+    """Serve the Dramatiq dashboard for staff users only."""
+    # Get the broker from django-dramatiq
+    broker = dramatiq.get_broker()
+
+    # Create the dashboard app
+    dashboard_app = dramatiq_dashboard.DashboardApp(broker=broker, prefix="")
+
+    # Create a WSGI environment
+    environ = {
+        "REQUEST_METHOD": request.method,
+        "PATH_INFO": request.path_info,
+        "QUERY_STRING": request.META.get("QUERY_STRING", ""),
+        "CONTENT_TYPE": request.META.get("CONTENT_TYPE", ""),
+        "CONTENT_LENGTH": request.META.get("CONTENT_LENGTH", ""),
+        "HTTP_HOST": request.META.get("HTTP_HOST", ""),
+        "SERVER_NAME": request.META.get("SERVER_NAME", ""),
+        "SERVER_PORT": request.META.get("SERVER_PORT", ""),
+        "wsgi.url_scheme": request.scheme,
+        "wsgi.input": request,
+        "wsgi.errors": request,
+        "wsgi.version": (1, 0),
+        "wsgi.multithread": True,
+        "wsgi.multiprocess": False,
+        "wsgi.run_once": False,
+    }
+
+    # Add HTTP headers
+    for key, value in request.META.items():
+        if key.startswith("HTTP_"):
+            environ[key] = value
+
+    # Call the dashboard app
+    def start_response(status, response_headers):  # pylint: disable=unused-argument
+        # This will be called by the WSGI app
+        pass
+
+    # Get the response from the dashboard
+    response_body = dashboard_app(environ, start_response)
+
+    # Create Django response
+    response = HttpResponse(b"".join(response_body))
+    response.status_code = 200
+    response["Content-Type"] = "text/html; charset=utf-8"
+
+    return response
+
+
+class CoreAdminSite(admin.AdminSite):
+    """Custom admin site with Dramatiq dashboard integration."""
+
+    def get_urls(self):
+        """Add Dramatiq dashboard URL to admin URLs."""
+        urls = super().get_urls()
+        custom_urls = [
+            path("dramatiq/", dramatiq_dashboard_view, name="dramatiq_dashboard"),
+        ]
+        return custom_urls + urls
+
+    def index(self, request, extra_context=None):
+        """Add Dramatiq dashboard link to admin index."""
+        extra_context = extra_context or {}
+        extra_context["dramatiq_dashboard_url"] = reverse("admin:dramatiq_dashboard")
+        return super().index(request, extra_context)
+
+
+# Create custom admin site instance
+admin_site = CoreAdminSite(name="admin")

src/backend/core/api/viewsets/import_message.py

@@ -43,7 +43,7 @@ class ImportViewSet(viewsets.ViewSet):
         request=ImportFileSerializer,
         responses={
             202: OpenApiResponse(
-                description="Import started. Returns Celery task ID for tracking.",
+                description="Import started. Returns task ID for tracking.",
                 response={
                     "type": "object",
                     "properties": {
@@ -96,7 +96,7 @@ def import_file(self, request):
         request=ImportIMAPSerializer,
         responses={
             202: OpenApiResponse(
-                description="IMAP import started. Returns Celery task ID for tracking the import progress.",
+                description="IMAP import started. Returns task ID for tracking the import progress.",
                 response={
                     "type": "object",
                     "properties": {

src/backend/core/api/viewsets/send.py

@@ -120,7 +120,7 @@ def post(self, request):
             )
 
         # Launch async task for sending the message
-        task = send_message_task.delay(str(message.id), must_archive=must_archive)
+        task = send_message_task.send(str(message.id), must_archive=must_archive)
 
         # --- Finalize ---
         # Message state should be updated by prepare_outbound_message/send_message
@@ -130,4 +130,4 @@ def post(self, request):
         # Update thread stats after un-drafting
         message.thread.update_stats()
 
-        return Response({"task_id": task.id}, status=status.HTTP_200_OK)
+        return Response({"task_id": task.message_id}, status=status.HTTP_200_OK)

src/backend/core/api/viewsets/task.py

@@ -1,9 +1,9 @@
-"""API ViewSet for Celery task status."""
+"""API ViewSet for asynchronous task statuses."""
 
 import logging
 
-from celery import states as celery_states
-from celery.result import AsyncResult
+import dramatiq
+from dramatiq.results import ResultFailure, ResultMissing, ResultTimeout
 from drf_spectacular.utils import (
     OpenApiExample,
     extend_schema,
@@ -14,10 +14,13 @@
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
-from messages.celery_app import app as celery_app
+from core.utils import get_task_progress
 
 logger = logging.getLogger(__name__)
 
+# Map Dramatiq states to Celery-like states for frontend compatibility
+DRAMATIQ_STATES = ["PENDING", "SUCCESS", "FAILURE", "RETRY", "REJECTED", "PROGRESS"]
+
 
 @extend_schema(
     tags=["tasks"],
@@ -34,9 +37,7 @@
         200: inline_serializer(
             name="TaskStatusResponse",
             fields={
-                "status": drf_serializers.ChoiceField(
-                    choices=sorted({*celery_states.ALL_STATES, "PROGRESS"})
-                ),
+                "status": drf_serializers.ChoiceField(choices=sorted(DRAMATIQ_STATES)),
                 "result": drf_serializers.JSONField(allow_null=True),
                 "error": drf_serializers.CharField(allow_null=True),
             },
@@ -59,38 +60,65 @@
     ],
 )
 class TaskDetailView(APIView):
-    """View to retrieve the status of a Celery task."""
+    """View to retrieve the status of a task."""
 
     permission_classes = [permissions.IsAuthenticated]
 
     def get(self, request, task_id):
-        """Get the status of a Celery task."""
-
-        task_result = AsyncResult(task_id, app=celery_app)
+        """Get the status of a task."""
+        # Try to fetch a result from Dramatiq's result backend
+        broker = dramatiq.get_broker()
+        result_backend = broker.get_results_backend()
 
-        # By default unknown tasks will be in PENDING. There is no reliable
-        # way to check if a task exists or not with Celery.
-        # https://github.com/celery/celery/issues/3596#issuecomment-262102185
+        if result_backend is not None:
+            try:
+                # Retrieve a Message for this task id from the backend, then get result
+                # See Dramatiq results API: message.get_result(...)
+                message = result_backend.get_message(task_id)
+                result = message.get_result(backend=result_backend, block=False)
+                return Response(
+                    {
+                        "status": "SUCCESS",
+                        "result": result,
+                        "error": None,
+                    }
+                )
+            except ResultMissing:
+                # No result yet; fall through to progress/pending logic
+                pass
+            except ResultFailure as exc:
+                return Response(
+                    {
+                        "status": "FAILURE",
+                        "result": None,
+                        "error": str(exc),
+                    }
+                )
+            except ResultTimeout as exc:
+                # Treat timeouts as pending
+                logger.debug("Result timeout for task %s: %s", task_id, exc)
 
-        # Prepare the response data
-        result_data = {
-            "status": task_result.status,
-            "result": None,
-            "error": None,
-        }
+        # Check if we have progress data for this task
+        progress_data = get_task_progress(task_id)
 
-        # If the result is a dict with status/result/error, unpack and propagate status
-        if isinstance(task_result.result, dict) and set(task_result.result.keys()) >= {
-            "status",
-            "result",
-            "error",
-        }:
-            result_data["status"] = task_result.result["status"]
-            result_data["result"] = task_result.result["result"]
-            result_data["error"] = task_result.result["error"]
-        else:
-            result_data["result"] = task_result.result
-        if task_result.state == "PROGRESS" and task_result.info:
-            result_data.update(task_result.info)
+        if progress_data:
+            # Task is in progress
+            return Response(
+                {
+                    "status": "PROGRESS",
+                    "result": None,
+                    "error": None,
+                    "progress": progress_data.get("progress"),
+                    "message": progress_data.get("metadata", {}).get("message"),
+                    "timestamp": progress_data.get("timestamp"),
+                }
+            )
 
-        return Response(result_data)
+        # Default to pending when no result and no progress
+        return Response(
+            {
+                "status": "PENDING",
+                "result": None,
+                "error": None,
+            }
+        )