Django: integrate Celery task queue to handle asynchronous tasks

Integrate Celery task queue. Redis serves as the message broker for the
Celery task queue. Redis can run in a seperate container with
of-the-shelf available images.

Gevent is used an efficient alternative compared to multiple python
processes, where each process has to load it's own runtime resulting in
high memory requirement. Celery is used for functions that wait for ReST
API calls to Leshan, an IO-bound task. Currently, Celery starts
1 worker (1 CPU), however, this worker is sufficient to handle more than
100 simultaneous ReST API calls efficiently.

Signed-off-by: Jonas Remmert <jremmert@gmx.net>

server/django/django_start.sh

@@ -25,5 +25,12 @@ else
     echo "lwm2m resource types data already loaded, skipping..."
 fi
 
+# Choosing gevent as the event loop for celery for significantly lower memory
+# usage. The application is IO bound as it is waiting for ReST API responses.
+#
+# -P gevent -c 100: 100 concurrent workers
+echo "Starting the Celery worker..."
+celery -A server worker --loglevel=info -P gevent -c 100 2>&1 | tee -a "logs/celery_$(date +%Y-%m-%d_%H-%M-%S).log" &
+
 echo "Starting the server..."
 python manage.py runserver 0.0.0.0:8000 2>&1 | tee -a "$logfile"

server/django/requirements.txt

@@ -3,6 +3,8 @@ django-extensions==3.2.3
 requests==2.31.0
 djangorestframework==3.15.1
 whitenoise==6.6.0
+celery[redis]==5.4.0
+gevent==24.2.1
 
 # Generation of Entity-Relationship Diagrams
 graphviz==0.20.3

server/django/server/__init__.py

@@ -0,0 +1,5 @@
+# This will make sure the app is always imported when
+# Django starts so that shared_task will use this app.
+from .celery import app as celery_app
+
+__all__ = ('celery_app',)

server/django/server/celery.py

@@ -0,0 +1,21 @@
+import os
+from celery import Celery
+
+# Set the default Django settings module for the 'celery' program.
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'server.settings')
+
+app = Celery('server')
+
+# Using a string here means the worker doesn't have to serialize
+# the configuration object to child processes.
+# - namespace='CELERY' means all celery-related configuration keys
+#   should have a `CELERY_` prefix.
+app.config_from_object('django.conf:settings', namespace='CELERY')
+
+# Load task modules from all registered Django apps.
+app.autodiscover_tasks()
+
+
+@app.task(bind=True, ignore_result=True)
+def debug_task(self):
+    print(f'Request: {self.request!r}')

server/django/server/settings.py

@@ -59,6 +59,10 @@
             'level': 'DEBUG',
             'propagate': False,
         },
+        'celery': {
+            'handlers': ['console'],
+            'level': 'INFO',
+        },
     },
 }
 
@@ -135,6 +139,15 @@
     }
 }
 
+# Celery asynchroneous task queue
+# Set Redis environment depending on on whether the application runs in a
+# container or native.
+CELERY_BROKER_URL = os.getenv('CELERY_BROKER_URL', 'redis://localhost:6379/0')
+CELERY_RESULT_BACKEND = os.getenv('CELERY_RESULT_BACKEND', 'redis://localhost:6379/0')
+CELERY_ACCEPT_CONTENT = ['json']
+CELERY_TASK_SERIALIZER = 'json'
+CELERY_RESULT_SERIALIZER = 'json'
+CELERY_TIMEZONE = 'UTC'
 
 # Password validation
 # https://docs.djangoproject.com/en/5.0/ref/settings/#auth-password-validators

server/docker-compose.yml

@@ -12,6 +12,15 @@ services:
       - mynetwork
     environment:
       - LESHAN_URI=http://leshan:8080
+      - CELERY_BROKER_URL=redis://redis:6379/0
+      - CELERY_RESULT_BACKEND=redis://redis:6379/0
+
+  redis:
+    image: redis:7.2.5
+    networks:
+      - mynetwork
+    ports:
+      - "6379:6379"
 
   leshan:
     build: