add celery; implement timed_completion in form admin

Dockerfile

@@ -7,27 +7,32 @@ COPY pyproject.toml poetry.lock /opt/services/djangoapp/src/
 
 # dependencies
 RUN apt-get update \
-    && apt-get install -y build-essential libpq-dev libqpdf-dev pip \
-                          python3-dev python3-importlib-metadata \
+    && apt-get install -y build-essential libpq-dev libqpdf-dev pip xz-utils \
+                          python3-dev python3-importlib-metadata wget redis \
     && apt-get install -y --no-install-recommends ffmpeg \
-    && rm -rf /var/lib/apt/lists/* \
     && pip install wheel poetry && poetry config virtualenvs.create false \
     && poetry install --no-dev --no-root -E reviewpanel \
+    && rm -rf /var/lib/apt/lists/* \
     && apt-get purge -y --auto-remove build-essential
 # virtualenvs.create option because we don't need an extra virtualenv here
 
-# copy the project code
 COPY . /opt/services/djangoapp/src
-
 # install the root package too
 RUN poetry install --no-dev
 
+COPY resources/s6-rc.d /etc/s6-overlay/s6-rc.d
+ARG S6_VERSION=3.1.0.1
+ARG S6_URL=https://github.com/just-containers/s6-overlay/releases/download
+RUN arch="$(dpkg --print-architecture)"; \
+    case "$arch" in arm64) s6arch='aarch64' ;; amd64) s6arch='amd64' ;; esac; \
+    wget -O s6.tar.xz ${S6_URL}/v${S6_VERSION}/s6-overlay-noarch.tar.xz; \
+    wget -O s6arch.tar.xz ${S6_URL}/v${S6_VERSION}/s6-overlay-$s6arch.tar.xz; \
+    tar -C / -Jxpf s6.tar.xz; \
+    tar -C / -Jxpf s6arch.tar.xz; \
+    rm s6.tar.xz s6arch.tar.xz
+
 EXPOSE 8000
 
-# TODO: change to an app user?
+ENTRYPOINT ["/init"]
 
-CMD (cd ../requirements; \
-    touch requirements.txt && pip install -r requirements.txt); \
-    python3 manage.py collectstatic --noinput && ( \
-    if [ "$AUTO_MIGRATE" != "" ]; then python3 manage.py migrate --noinput; fi \
-    ) && exec gunicorn -c config/gunicorn.conf.py config.wsgi:application
+# TODO: change to an app user?

config/celery.py

@@ -0,0 +1,9 @@
+import os
+from celery import Celery
+
+
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'settings')
+
+app = Celery('formative')
+app.config_from_object('django.conf:settings', namespace='CELERY')
+app.autodiscover_tasks()

config/gunicorn.conf.py

@@ -1,5 +1,5 @@
 import multiprocessing
 
 bind = ':8000'
-timeout = 180
+timeout = 45
 workers = multiprocessing.cpu_count() * 2 + 1

config/wsgi.py

@@ -10,6 +10,7 @@
 import os
 
 from django.core.wsgi import get_wsgi_application
+from .celery import app as celery_app
 
 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'settings')
 

docker-compose.yml

@@ -44,9 +44,6 @@ services:
     volumes:
       - database_volume:/var/lib/postgresql/data
 
-  redis:
-    image: redis:6
-
   postfix:
     image: juanluisbaptiste/postfix:1.3
     env_file:

formative/admin/actions.py

@@ -1,21 +1,20 @@
 from django.conf import settings
 from django.contrib import admin, auth, messages
 from django.contrib.auth.models import User
-from django.core import mail
 from django.core.exceptions import ValidationError
 from django.db import IntegrityError
 from django.db.models import Count, Max
 from django.http import HttpResponseRedirect
-from django.template import Template
 from django.template.response import TemplateResponse
 from django.urls import path
 from django.utils.text import capfirst
-import time, csv, io
+import csv, io
 
 from ..forms import MoveBlocksAdminForm, EmailAdminForm, FormPluginsAdminForm, \
     UserImportForm, ExportAdminForm
 from ..models import Form, FormBlock, SubmissionRecord
-from ..utils import send_email, submission_link, TabularExport
+from ..tasks import send_email_for_submissions
+from ..utils import TabularExport
 
 
 class UserActionsMixin:
@@ -201,44 +200,15 @@ def move_blocks_action(self, request, queryset):
 
 
 class SubmissionActionsMixin:
-    EMAILS_PER_SECOND = 10
-
     @admin.action(description='Send an email to applicants')
     def send_email(self, request, queryset):
         if '_send' in request.POST:
-            # TODO: celery task
-            subject = Template(request.POST['subject'])
-            content = Template(request.POST['content'])
-
-            iterator = queryset.iterator()
-            batch, form, last_time, done, n = [], None, None, False, 0
-            while not done:
-                submission = next(iterator, None)
-                if not submission: done = True
-                else: batch.append(submission)
-
-                if len(batch) == self.EMAILS_PER_SECOND or done:
-                    if last_time:
-                        this_time = time.time()
-                        remaining = last_time + 1 - this_time
-                        if remaining > 0: time.sleep(remaining)
-                    last_time = time.time()
-
-                    with mail.get_connection() as conn:
-                        for sub in batch:
-                            if not form: form = sub._get_form()
-                            context = {
-                                'submission': sub, 'form': form,
-                                'submission_link': submission_link(sub, form)
-                            }
-                            if sub._submitted: sub._update_context(form,
-                                                                   context)
-                            n += 1
-                            send_email(content, sub._email, subject,
-                                       context=context, connection=conn)
-                    batch = []
-
-            msg = f'Emails sent to {n} recipients.'
+            send_email_for_submissions.delay(
+                queryset.model._meta.model_name,
+                list(queryset.values_list('pk', flat=True)),
+                request.POST['subject'], request.POST['content']
+            )
+            msg = f'Email sending started for {queryset.count()} recipients.'
             self.message_user(request, msg, messages.SUCCESS)
 
             return HttpResponseRedirect(request.get_full_path())

formative/admin/formative.py

@@ -27,6 +27,7 @@
 from ..plugins import get_matching_plugin
 from ..signals import register_program_settings, register_form_settings, \
     register_user_actions, form_published_changed
+from ..tasks import timed_complete_form
 from ..utils import submission_link
 from .actions import UserActionsMixin, FormActionsMixin,FormBlockActionsMixin, \
     SubmissionActionsMixin
@@ -187,6 +188,10 @@ def save_form(self, request, form, change):
             if 'emails' in obj.options and not obj.options['emails']:
                 del obj.options['emails']
 
+        if 'timed_completion' in form.changed_data:
+            val = form.cleaned_data['timed_completion']
+            timed_complete_form.apply_async(args=(obj.id, val), eta=val)
+
         return obj
 
     def response_post_save_change(self, request, obj):

formative/tasks.py

@@ -0,0 +1,59 @@
+from django.apps import apps
+from django.core import mail
+from django.template import Template
+from django.utils import timezone
+from celery import shared_task
+import time
+
+from .models import Form
+from .utils import send_email, submission_link
+
+
+EMAILS_PER_SECOND = 10
+
+@shared_task
+def send_email_for_submissions(model_name, id_values, subject_str, content_str):
+    model = apps.get_model(f'formative.{model_name}')
+    queryset = model.objects.filter(pk__in=id_values)
+
+    subject, content = Template(subject_str), Template(content_str)
+    iterator = queryset.iterator()
+    batch, form, last_time, done, n = [], None, None, False, 0
+    while not done:
+        submission = next(iterator, None)
+        if not submission: done = True
+        else: batch.append(submission)
+
+        if len(batch) == EMAILS_PER_SECOND or done:
+            if last_time:
+                this_time = time.time()
+                remaining = last_time + 1 - this_time
+                if remaining > 0: time.sleep(remaining)
+            last_time = time.time()
+
+            with mail.get_connection() as conn:
+                for sub in batch:
+                    if not form: form = sub._get_form()
+                    context = {
+                        'submission': sub, 'form': form,
+                        'submission_link': submission_link(sub, form)
+                    }
+                    if sub._submitted: sub._update_context(form, context)
+                    n += 1
+                    send_email(content, sub._email, subject,
+                               context=context, connection=conn)
+            batch = []
+    return n
+
+@shared_task
+def timed_complete_form(form_id, datetime_val):
+    try: form = Form.objects.get(id=form_id)
+    except Form.DoesNotExist: return
+
+    if form.timed_completion() != datetime_val:
+        return False # it gets rescheduled on value change, so don't do anything
+
+    form.status = Form.Status.COMPLETED
+    form.completed = timezone.now()
+    form.save()
+    return True

formative/utils.py

@@ -1,4 +1,3 @@
-from django.apps import apps
 from django.db.models import Model, Q, OuterRef, Max, Count
 from django.conf import settings
 from django.core import mail

manage.py

@@ -3,6 +3,7 @@
 import os
 import sys
 
+from config.celery import app as celery_app
 
 def main():
     """Run administrative tasks."""