feat: add celery

Author: sudip-khanal

.pre-commit-config.yaml

@@ -52,11 +52,11 @@ repos:
     hooks:
       # Linter
       - id: ruff
-        types_or: [python, pyi, jupyter, toml]
+        types_or: [python, pyi, jupyter, pyproject]
         args: [--fix, --exit-non-zero-on-fix]
       # Formatter
       - id: ruff-format
-        types_or: [python, pyi, jupyter, toml]
+        types_or: [python, pyi, jupyter, pyproject]
 
   - repo: https://github.com/RobertCraigie/pyright-python
     rev: v1.1.398

apps/common/management/commands/run_celery_dev.py

@@ -0,0 +1,26 @@
+import shlex
+import subprocess
+import typing
+from pathlib import Path
+
+from django.core.management.base import BaseCommand
+from django.utils import autoreload
+
+WORKER_STATE_DIR = Path("/var/run/celery")
+
+CMD = "celery -A main worker -E --concurrency=2 -l info"
+
+
+def restart_celery(*args: typing.Any, **kwargs: typing.Any):
+    kill_worker_cmd = "pkill -9 celery"
+    subprocess.call(shlex.split(kill_worker_cmd))  # noqa: S603
+    subprocess.call(shlex.split(CMD))  # noqa: S603
+
+
+class Command(BaseCommand):
+    @typing.override
+    def handle(self, *args: typing.Any, **options: typing.Any):
+        self.stdout.write("Starting celery worker with autoreload...")
+        if not Path.exists(WORKER_STATE_DIR):
+            Path.mkdir(WORKER_STATE_DIR, parents=True)
+        autoreload.run_with_reloader(restart_celery, args=None, kwargs=None)

apps/common/management/commands/wait_for_resources.py

@@ -5,9 +5,11 @@
 
 import requests
 from django.conf import settings
+from django.core.cache import cache
 from django.core.management.base import BaseCommand, CommandParser
 from django.db import connections
 from django.db.utils import OperationalError
+from redis.exceptions import ConnectionError as RedisConnectionError
 
 
 class TimeoutException(Exception): ...
@@ -37,6 +39,25 @@ def wait_for_db(self):
 
         self.stdout.write(self.style.SUCCESS(f"DB is available after {time.time() - start_time} seconds"))
 
+    def wait_for_redis(self):
+        self.stdout.write("Waiting for Redis...")
+        redis_conn = None
+        start_time = time.time()
+        while True:
+            try:
+                cache.set("wait-for-it-ping", "pong", timeout=1)  # Set a key to check Redis availability
+                redis_conn = cache.get("wait-for-it-ping")  # Try to get the value back from Redis
+                if redis_conn != "pong":
+                    raise TypeError
+                break
+            except (RedisConnectionError, TypeError):
+                ...
+            # Try again
+            self.stdout.write(self.style.WARNING("Redis not available, waiting..."))
+            time.sleep(1)
+
+        self.stdout.write(self.style.SUCCESS(f"Redis is available after {time.time() - start_time} seconds"))
+
     def wait_for_minio(self):
         self.stdout.write("Waiting for Minio...")
         AWS_S3_CONFIG_OPTIONS = getattr(settings, "AWS_S3_CONFIG_OPTIONS", None) or {}
@@ -69,6 +90,8 @@ def add_arguments(self, parser: CommandParser):
             help="The maximum time (in seconds) the command is allowed to run before timing out. Default is 10 min.",
         )
         parser.add_argument("--db", action="store_true", help="Wait for DB to be available")
+        parser.add_argument("--celery-queue", action="store_true", help="Wait for Celery queue to be available")
+        parser.add_argument("--redis", action="store_true", help="Wait for Redis to be available")
         parser.add_argument("--minio", action="store_true", help="Wait for MinIO (S3) storage to be available")
         parser.add_argument("--all", action="store_true", help="Wait for all to be available")
 
@@ -86,6 +109,10 @@ def handle(self, **kwargs: typing.Any):
                 self.wait_for_db()
             if _all or kwargs["minio"]:
                 self.wait_for_minio()
+            if _all or kwargs["redis"]:
+                self.wait_for_redis()
+            if _all or kwargs["celery_queue"]:
+                self.wait_for_redis()
         except TimeoutException:
             ...
         finally:

docker-compose.yaml

@@ -42,6 +42,15 @@ x-server: &base_server_setup
     - ipython_data_local:/root/.ipython/profile_default   # persist ipython data, including ipython history
   depends_on:
     - db
+    - redis
+    - mailpit
+
+x-worker: &base_worker_setup
+  <<: *base_server_setup
+  restart: unless-stopped
+  environment:
+    <<: *base_server_environments
+    APP_TYPE: "WORKER"
 
 services:
   db:
@@ -53,12 +62,17 @@ services:
     volumes:
       - postgres_data17:/var/lib/postgresql/data
 
+  redis:
+    image: redis:8
+    volumes:
+      - redis_data:/data
+
   mailpit:
     image: axllent/mailpit
     container_name: mailpit
     restart: unless-stopped
     volumes:
-      - mailpit-data:/data
+      - mailpit_data:/data
     ports:
         - 8025:8025
     environment:
@@ -77,8 +91,20 @@ services:
       - 127.0.0.1:8000:8000
     depends_on:
       - db
+      - redis
+
+  worker:
+    <<: *base_worker_setup
+    command: bash -c "/code/misc/dev/run_worker.sh"
+    healthcheck:
+      test: ["CMD-SHELL", "celery -A main inspect ping -d celery@$$HOSTNAME || exit 1"]
+      interval: 30s
+      timeout: 5s
+      retries: 3
+      start_period: 30s
 
 volumes:
   postgres_data17:
   ipython_data_local:
-  mailpit-data:
+  mailpit_data:
+  redis_data:

main/__init__.py

@@ -1,3 +1,7 @@
 import django_stubs_ext
 
+from .celery import app as celery_app
+
+__all__ = ("celery_app",)
+
 django_stubs_ext.monkeypatch()

main/celery.py

@@ -0,0 +1,20 @@
+import logging
+import os
+
+from celery import Celery
+
+logger = logging.getLogger(__name__)
+
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "main.settings")
+
+app = Celery("main")
+
+app.config_from_object("django.conf:settings", namespace="CELERY")
+
+
+app.autodiscover_tasks()
+
+
+@app.task(bind=True)  # type: ignore[reportIncompatibleVariableOverride]
+def debug_task(self):
+    logger.info("Request: %s", self.request)

main/settings.py

@@ -59,6 +59,11 @@
     DEFAULT_FROM_EMAIL=(str, None),
     EMAIL_TO=(str, None),
     EMAIL_USE_TLS=(bool, False),
+    # celery
+    CELERY_REDIS_URL=str,  # redis://redis:6379/0
+    # Cache
+    CACHE_REDIS_URL=str,  # redis://redis:6379/1
+    TEST_CACHE_REDIS_URL=(str, None),  # redis://redis:6379/11
 )
 
 
@@ -405,3 +410,9 @@
 EMAIL_HOST_PASSWORD = env("EMAIL_HOST_PASSWORD")
 DEFAULT_FROM_EMAIL = env("DEFAULT_FROM_EMAIL")
 EMAIL_TO = env("EMAIL_TO")
+
+# Celery
+CELERY_REDIS_URL = env("CELERY_REDIS_URL")
+CACHE_REDIS_URL = env("CACHE_REDIS_URL")
+CELERY_BROKER_URL = CELERY_REDIS_URL
+CELERY_RESULT_BACKEND = CELERY_REDIS_URL

misc/dev/run_worker.sh

@@ -0,0 +1,7 @@
+#!/bin/bash -e
+
+REDIS_HOST_PORT=$(echo $CELERY_REDIS_URL | sed 's|redis://\([^/]*\)/.*|\1|')
+
+./manage.py wait_for_resources --db --celery-queue
+
+./manage.py run_celery_dev

pyproject.toml

@@ -32,6 +32,7 @@ dependencies = [
     "django-mdeditor>=0.1.20",
     "django-simple-captcha>=0.6.3",
     "django-stubs>=5.1.3",
+    "celery>=5.6.3",
 ]
 
 [dependency-groups]
@@ -62,7 +63,6 @@ extend-exclude = [
     "venv",
     "**/migrations/*",
     "__pycache__",
-    "uv.lock"
 ]
 
 [tool.ruff.lint]