Add UserApprovalTask

Based on https://docs.wagtail.org/en/stable/extending/custom_tasks.html

bakerydemo/base/migrations/0019_userapprovaltask.py

@@ -0,0 +1,59 @@
+# Generated by Django 4.2.dev20221115094701 on 2022-11-18 11:04
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("wagtailcore", "0078_referenceindex"),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ("base", "0018_add_genericsettings_and_sitesettings"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="UserApprovalTaskState",
+            fields=[
+                (
+                    "taskstate_ptr",
+                    models.OneToOneField(
+                        auto_created=True,
+                        on_delete=django.db.models.deletion.CASCADE,
+                        parent_link=True,
+                        primary_key=True,
+                        serialize=False,
+                        to="wagtailcore.taskstate",
+                    ),
+                ),
+            ],
+            bases=("wagtailcore.taskstate",),
+        ),
+        migrations.CreateModel(
+            name="UserApprovalTask",
+            fields=[
+                (
+                    "task_ptr",
+                    models.OneToOneField(
+                        auto_created=True,
+                        on_delete=django.db.models.deletion.CASCADE,
+                        parent_link=True,
+                        primary_key=True,
+                        serialize=False,
+                        to="wagtailcore.task",
+                    ),
+                ),
+                (
+                    "user",
+                    models.ForeignKey(
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        to=settings.AUTH_USER_MODEL,
+                    ),
+                ),
+            ],
+            bases=("wagtailcore.task",),
+        ),
+    ]

bakerydemo/base/models.py

@@ -1,5 +1,6 @@
 from __future__ import unicode_literals
 
+from django.conf import settings
 from django.contrib.contenttypes.fields import GenericRelation
 from django.db import models
 from django.utils.translation import gettext as _
@@ -26,6 +27,8 @@ from wagtail.models import (
     Page,
     PreviewableMixin,
     RevisionMixin,
+    Task,
+    TaskState,
     TranslatableMixin,
     WorkflowMixin,
 )
@@ -510,3 +513,58 @@ class SiteSettings(BaseSiteSetting):
     panels = [
         FieldPanel("title_suffix"),
     ]
+
+
+class UserApprovalTaskState(TaskState):
+    pass
+
+
+class UserApprovalTask(Task):
+    user = models.ForeignKey(
+        settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=False
+    )
+
+    admin_form_fields = Task.admin_form_fields + ["user"]
+
+    task_state_class = UserApprovalTaskState
+
+    # prevent editing of `user` after the task is created
+    # by default, this attribute contains the 'name' field to prevent tasks from being renamed
+    admin_form_readonly_on_edit_fields = Task.admin_form_readonly_on_edit_fields + [
+        "user"
+    ]
+
+    def user_can_access_editor(self, page, user):
+        return user == self.user
+
+    def page_locked_for_user(self, page, user):
+        return user != self.user
+
+    def get_actions(self, page, user):
+        if user == self.user:
+            return [
+                ("approve", "Approve", False),
+                ("reject", "Reject", False),
+                ("cancel", "Cancel", False),
+            ]
+        else:
+            return []
+
+    def on_action(self, task_state, user, action_name, **kwargs):
+        if action_name == "cancel":
+            return task_state.workflow_state.cancel(user=user)
+        else:
+            return super().on_action(task_state, user, action_name, **kwargs)
+
+    def get_task_states_user_can_moderate(self, user, **kwargs):
+        if user == self.user:
+            # get all task states linked to the (base class of) current task
+            return TaskState.objects.filter(
+                status=TaskState.STATUS_IN_PROGRESS, task=self.task_ptr
+            )
+        else:
+            return TaskState.objects.none()
+
+    @classmethod
+    def get_description(cls):
+        return _("Only a specific user can approve this task")