Remove and migrate more celery tasks (#4742)

These have now been migrated to `lambda_tasks` and the periodic tasks
set up.

See #4408

Author: jmsmkn

app/config/settings.py

@@ -1229,18 +1229,6 @@ def get_private_ip():
         "task": "grandchallenge.algorithms.tasks.update_associated_challenges",
         "schedule": crontab(hour=3, minute=0),
     },
-    "send_new_unread_direct_messages_emails": {
-        "task": "grandchallenge.direct_messages.tasks.send_new_unread_direct_messages_emails",
-        "schedule": crontab(hour=3, minute=30),
-    },
-    "send_unread_notification_emails": {
-        "task": "grandchallenge.notifications.tasks.send_unread_notification_emails",
-        "schedule": crontab(hour=4, minute=0),
-    },
-    "update_site_statistics": {
-        "task": "grandchallenge.statistics.tasks.update_site_statistics_cache",
-        "schedule": crontab(hour=5, minute=30),
-    },
     "send_onboarding_task_reminder_emails": {
         "task": "grandchallenge.challenges.tasks.send_onboarding_task_reminder_emails",
         "schedule": crontab(day_of_week="mon", hour=6, minute=0),
@@ -1293,10 +1281,6 @@ def get_private_ip():
         "task": "grandchallenge.emails.tasks.cleanup_sent_raw_emails",
         "schedule": timedelta(hours=1),
     },
-    "cleanup_celery_backend": {
-        "task": "grandchallenge.core.tasks.cleanup_celery_backend",
-        "schedule": timedelta(hours=1),
-    },
     "logout_privileged_users": {
         "task": "grandchallenge.browser_sessions.tasks.logout_privileged_users",
         "schedule": timedelta(hours=1),

app/grandchallenge/cases/tasks.py

@@ -583,16 +583,6 @@ def handle_dicom_import_error(
     )
 
 
-@acks_late_micro_short_task(
-    name=f"{__name__}.handle_health_imaging_import_job_event",
-    retry_on=(LockNotAcquiredException, RetryStep),
-)
-@transaction.atomic
-def handle_health_imaging_import_job_event_celery(*, event):
-    # TODO: 4408 Remove, this is still here to handle existing tasks on SQS
-    return handle_health_imaging_import_job_event(event=event)
-
-
 @lambda_task(retry_on=(LockNotAcquiredException, RetryStep))
 def handle_health_imaging_import_job_event(*, event: dict):
     job_name = event["jobName"]

app/grandchallenge/codebuild/tasks.py

@@ -35,15 +35,6 @@ def create_codebuild_build(*, pk):
     Build.objects.create(webhook_message=ghwm, algorithm_image=algorithm_image)
 
 
-@acks_late_micro_short_task(name=f"{__name__}.handle_completed_build_event")
-@transaction.atomic
-def handle_completed_build_event_celery(*, build_arn, build_status):
-    # TODO: 4408 Remove, this is still here to handle existing tasks on SQS
-    return handle_completed_build_event(
-        build_arn=build_arn, build_status=build_status
-    )
-
-
 @lambda_task
 def handle_completed_build_event(*, build_arn: str, build_status: str):
     from grandchallenge.codebuild.models import Build

app/grandchallenge/core/tasks.py

@@ -1,9 +1,6 @@
 from datetime import timedelta
 
 import boto3
-from billiard.exceptions import (
-    SoftTimeLimitExceeded as CelerySoftTimeLimitExceeded,
-)
 from django.conf import settings
 from django.contrib.sites.models import Site
 from django.db import transaction
@@ -25,8 +22,14 @@
 from grandchallenge.workstations.models import Session
 
 
-@acks_late_micro_short_task
+@acks_late_micro_short_task(name=f"{__name__}.cleanup_celery_backend")
 @transaction.atomic
+def cleanup_celery_backend_celery():
+    # TODO: 4408 Remove, this is still here to handle existing tasks on SQS
+    return cleanup_celery_backend()
+
+
+@lambda_task
 def cleanup_celery_backend():
     """Cleanup the Celery backend."""
     TaskResult.objects.filter(date_created__lt=now() - timedelta(days=7)).only(
@@ -37,23 +40,6 @@ def cleanup_celery_backend():
 CLOUDWATCH_METRICS_LIMIT = 1000
 
 
-@acks_late_micro_short_task(
-    name=f"{__name__}.put_cloudwatch_metrics",
-    ignore_result=True,
-    singleton=True,
-    # No need to retry here as the periodic task call this again
-    ignore_errors=(
-        LockError,
-        CelerySoftTimeLimitExceeded,
-        SoftTimeLimitExceeded,
-    ),
-)
-@transaction.atomic
-def put_cloudwatch_metrics_celery():
-    # TODO: 4408 Remove, this is still here to handle existing tasks on SQS
-    return put_cloudwatch_metrics()
-
-
 @lambda_task(
     singleton=True,
     # No need to retry here as the periodic task call this again

app/grandchallenge/direct_messages/tasks.py

@@ -1,6 +1,7 @@
 from django.contrib.auth import get_user_model
 from django.contrib.sites.models import Site
 from django.db.models import Count, F, Q
+from lambda_tasks.decorators import lambda_task
 
 from grandchallenge.core.celery import acks_late_micro_short_task
 from grandchallenge.profiles.models import NotificationEmailOptions
@@ -47,7 +48,15 @@ def get_new_senders(*, user):
     return sorted(list(new_senders), key=lambda s: s.pk)
 
 
-@acks_late_micro_short_task
+@acks_late_micro_short_task(
+    name=f"{__name__}.send_new_unread_direct_messages_emails"
+)
+def send_new_unread_direct_messages_emails_celery():
+    # TODO: 4408 Remove, this is still here to handle existing tasks on SQS
+    return send_new_unread_direct_messages_emails()
+
+
+@lambda_task
 def send_new_unread_direct_messages_emails():
     site = Site.objects.get_current()
 

app/grandchallenge/discussion_forums/models.py

@@ -5,7 +5,6 @@
 from django.contrib.auth.models import AnonymousUser
 from django.core.exceptions import ObjectDoesNotExist, ValidationError
 from django.db import models
-from django.db.transaction import on_commit
 from django_extensions.db.fields import AutoSlugField
 from guardian.shortcuts import assign_perm, remove_perm
 from guardian.utils import get_anonymous_user
@@ -123,14 +122,10 @@ def save(self, *args, **kwargs):
         if adding:
             self.assign_permissions()
             self.last_post_on = self.created
-            on_commit(
-                create_forum_notifications.signature(
-                    kwargs={
-                        "object_pk": self.pk,
-                        "app_label": self._meta.app_label,
-                        "model_name": self._meta.object_name,
-                    }
-                ).apply_async
+            create_forum_notifications.execute_on_commit(
+                object_pk=self.pk,
+                app_label=self._meta.app_label,
+                model_name=self._meta.object_name,
             )
 
         if self.has_changed("is_locked"):
@@ -264,14 +259,10 @@ def save(self, *args, **kwargs):
             self.assign_permissions()
             self.topic.mark_as_read(user=self.creator)
             if not self.is_alone:
-                on_commit(
-                    create_forum_notifications.signature(
-                        kwargs={
-                            "object_pk": self.pk,
-                            "app_label": self._meta.app_label,
-                            "model_name": self._meta.object_name,
-                        }
-                    ).apply_async
+                create_forum_notifications.execute_on_commit(
+                    object_pk=self.pk,
+                    app_label=self._meta.app_label,
+                    model_name=self._meta.object_name,
                 )
 
         self.topic.last_post = self

app/grandchallenge/discussion_forums/tasks.py

@@ -1,7 +1,10 @@
+from uuid import UUID
+
 from actstream.actions import follow
 from celery.utils.log import get_task_logger
 from django.apps import apps
 from django.db import transaction
+from lambda_tasks.decorators import lambda_task
 
 from grandchallenge.core.celery import acks_late_micro_short_task
 from grandchallenge.core.exceptions import LockNotAcquiredException
@@ -13,9 +16,20 @@
 logger = get_task_logger(__name__)
 
 
-@acks_late_micro_short_task(retry_on=(LockNotAcquiredException,))
+@acks_late_micro_short_task(
+    name=f"{__name__}.create_forum_notifications",
+    retry_on=(LockNotAcquiredException,),
+)
 @transaction.atomic
-def create_forum_notifications(*, object_pk, app_label, model_name):
+def create_forum_notifications_celery(**kwargs):
+    # TODO: 4408 Remove, this is still here to handle existing tasks on SQS
+    return create_forum_notifications(**kwargs)
+
+
+@lambda_task(retry_on=(LockNotAcquiredException,))
+def create_forum_notifications(
+    *, object_pk: str | UUID, app_label: str, model_name: str
+):
     from grandchallenge.discussion_forums.models import (
         ForumPost,
         ForumTopic,

app/grandchallenge/notifications/tasks.py

@@ -1,5 +1,6 @@
 from django.contrib.sites.models import Site
 from django.db.models import Count, F, Q
+from lambda_tasks.decorators import lambda_task
 
 from grandchallenge.core.celery import acks_late_micro_short_task
 from grandchallenge.profiles.models import (
@@ -8,7 +9,13 @@
 )
 
 
-@acks_late_micro_short_task
+@acks_late_micro_short_task(name=f"{__name__}.send_unread_notification_emails")
+def send_unread_notification_emails_celery():
+    # TODO: 4408 Remove, this is still here to handle existing tasks on SQS
+    return send_unread_notification_emails()
+
+
+@lambda_task
 def send_unread_notification_emails():
     site = Site.objects.get_current()
 

app/grandchallenge/statistics/tasks.py

@@ -4,6 +4,7 @@
 from django.core.cache import cache
 from django.db.models import Count, Sum
 from django.utils.timezone import now
+from lambda_tasks.decorators import lambda_task
 
 from grandchallenge.algorithms.models import (
     Algorithm,
@@ -29,7 +30,13 @@
 from grandchallenge.workstations.models import Session, WorkstationImage
 
 
-@acks_late_micro_short_task
+@acks_late_micro_short_task(name=f"{__name__}.update_site_statistics_cache")
+def update_site_statistics_cache_celery():
+    # TODO: 4408 Remove, this is still here to handle existing tasks on SQS
+    return update_site_statistics_cache()
+
+
+@lambda_task
 def update_site_statistics_cache():
     public_challenges = Challenge.objects.filter(hidden=False)
 

app/tests/challenges_tests/test_models.py

@@ -94,8 +94,7 @@ def test_participants_follow_forum(group):
 def test_non_posters_notified(
     group, settings, django_capture_on_commit_callbacks
 ):
-    settings.CELERY_TASK_ALWAYS_EAGER = True
-    settings.CELERY_TASK_EAGER_PROPAGATES = True
+    settings.LAMBDA_TASKS_EAGER = True
 
     p = UserFactory()
     u = UserFactory()