changed celery default queue, cleaned up core views, fixed celery queue name in local development

Author: briancaffey

.env.template

@@ -18,8 +18,8 @@ GOOGLE_OAUTH2_SECRET=abcde-fghij
 SOCIAL_AUTH_FACEBOOK_KEY = ''
 SOCIAL_AUTH_FACEBOOK_SECRET = ''
 
-CELERY_BROKER_URL=redis://redis:6379
-CELERY_RESULT_BACKEND=redis://redis:6379
+CELERY_BROKER_URL=redis://redis:6379/0
+CELERY_RESULT_BACKEND=redis://redis:6379/1
 
 DJANGO_EMAIL_HOST=mailhog
 DJANGO_EMAIL_PORT=1025

awscdk/awscdk/celery_default.py

@@ -4,7 +4,6 @@
     aws_ecs as ecs,
     aws_logs as logs,
     aws_cloudformation as cloudformation,
-    aws_sqs as sqs,
 )
 
 

backend/apps/core/management/commands/__init__.py

@@ -6,21 +6,32 @@
 https://avilpage.com/2017/05/how-to-auto-reload-celery-workers-in-development.html
 """
 
+import os
 import shlex
 import subprocess
 
 from django.core.management.base import BaseCommand
 from django.utils import autoreload
 
 
-def restart_celery():
-    cmd = "pkill -9 celery"
+def restart_celery(queue=None, concurrency=None):
+    cmd = 'pkill -9 celery'
     subprocess.call(shlex.split(cmd))
-    cmd = "celery worker --app=backend.celery_app:app --loglevel=info --concurrency=2 --max-memory-per-child=150000"  # noqa
+    cmd = f"celery worker --app=backend.celery_app:app --loglevel=info -Q {queue} -n worker-{queue}@%h --concurrency={os.environ.get('CONCURRENT_WORKERS', 2)} --max-memory-per-child=150000"  # noqa
     subprocess.call(shlex.split(cmd))
 
 
 class Command(BaseCommand):
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '-q', '--queue', nargs=1, default='celery', type=str
+        )
+        parser.add_argument('-c', '--concurrency', type=str)
+
     def handle(self, *args, **options):
-        print("Starting celery worker with autoreload...")
-        autoreload.run_with_reloader(restart_celery)
+        queue = options['queue'][0]
+        concurrency = options['concurrency'] or 1
+        print('Starting celery worker with autoreload...')
+        autoreload.run_with_reloader(
+            restart_celery, queue=queue, concurrency=concurrency
+        )

backend/apps/core/management/commands/watch_celery.py

@@ -15,7 +15,6 @@ class BaseTask(celery.Task):
 @task(bind=True, base=BaseTask)
 def debug_task(self):
     time.sleep(10)
-    print("Task is done")
 
 
 @periodic_task(
@@ -30,8 +29,8 @@ def debug_periodic_task():
 @task(bind=True, base=BaseTask)
 def send_test_email_task(self):
     send_mail(
-        "Subject here",
-        "Here is the message.",
+        "Email subject",
+        "Email message.",
         "from@example.com",
         ["to@example.com"],
         fail_silently=False,
@@ -40,6 +39,5 @@ def send_test_email_task(self):
 
 @task(bind=True, base=BaseTask)
 def sleep_task(self, seconds):
-    print("sleeping")
     time.sleep(int(seconds))
     return f"Slept {seconds} seconds"

backend/apps/core/tasks.py

@@ -3,9 +3,7 @@
 from . import views
 
 urlpatterns = [
-    path("", views.home, name="home"),
-    path("hello-world", views.hello_world, name="hello-world"),
-    path("debug-task/", views.debug_task_view, name="debug-task",),
+    path("health-check/", views.health_check, name="health-check"),
     path("celery/sleep-task/", views.sleep_task_view, name="sleep-task"),
     path(
         "debug/send-test-email/",

backend/apps/core/urls.py

@@ -18,54 +18,32 @@ def get(self, request):
         value = r.get("cached_value")
 
         if value:
-            print(value)
             count = value
         return JsonResponse({"count": count})
 
     def post(self, request):
         new_count = int(request.data["count"])
-        print("setting value to cache:")
-        print(new_count)
         r.set("cached_value", new_count)
         new_count = r.get("cached_value")
-        print("value from cache is...")
-        print(new_count)
         return JsonResponse({"count": new_count})
 
     def delete(self, request):
         r.delete("cached_value")
         return JsonResponse({"count": r.get("cached_value")})
 
 
-def hello_world(request):
-    response = JsonResponse(
-        {
-            "message": "Hello, World!",
-            "git_sha": os.environ.get("GIT_SHA", "<git SHA>"),
-            "debug": settings.DEBUG,
-            "format": "JSON",
-            "ssm_param": os.environ.get("MY_PARAM", "param_value"),
-        }
-    )
-    return response
-
-
-def home(request):
-    response = JsonResponse({"message": "Root"})
+def health_check(request):
+    response = JsonResponse({"message": "OK"})
     return response
 
 
-def debug_task_view(request):
-    debug_task.delay()
-    return JsonResponse({"message": "Task sent to queue."})
-
-
 @api_view(["POST"])
 def sleep_task_view(request):
     sleep_seconds = request.data.get("seconds")
-    print(sleep_seconds)
-    sleep_task.delay(sleep_seconds)
-    return JsonResponse({"message": "Sleep task submitted"})
+    sleep_task.apply_async(
+        [sleep_seconds], queue=settings.CELERY_QUEUE_DEFAULT
+    )
+    return JsonResponse({"message": f"Sleep task submitted ({sleep_seconds})"})
 
 
 def send_test_email(request):

backend/apps/core/views.py

@@ -12,6 +12,7 @@
 
 import os
 
+from kombu import Queue
 import redis
 
 # Build paths inside the project like this: os.path.join(BASE_DIR, ...)
@@ -202,8 +203,12 @@
 CELERY_ACCEPT_CONTENT = ["application/json"]
 CELERY_TASK_SERIALIZER = "json"
 CELERY_RESULT_SERIALIZER = "json"
-CELERY_BROKER_URL = f"redis://{REDIS_SERVICE_HOST}:6379/0"  # noqa
-CELERY_RESULT_BACKEND = f"redis://{REDIS_SERVICE_HOST}:6379/0"  # noqa
+CELERY_BROKER_URL = f"redis://{REDIS_SERVICE_HOST}:6379/1"  # noqa
+CELERY_RESULT_BACKEND = f"redis://{REDIS_SERVICE_HOST}:6379/2"  # noqa
+
+CELERY_QUEUE_DEFAULT = 'default'
+
+CELERY_QUEUES = (Queue(CELERY_QUEUE_DEFAULT, routing_key='default'),)
 
 
 AUTH_USER_MODEL = "accounts.CustomUser"

backend/backend/settings/base.py

@@ -128,11 +128,8 @@ services:
 
   celery:
     <<: *backend
-    build:
-      context: ./backend
-      dockerfile: scripts/dev/Dockerfile
     container_name: celery
-    command: bash -c 'python3 manage.py watch_celery'
+    command: bash -c 'python3 manage.py watch_celery --queue default --concurrency=2'
     volumes:
       - ./backend:/code
     ports: []