Branching Workflows based on value of specified Page field

Question

I have a DailyReflectionPage Model with a reflection_date field that forms the basis for the Page's slug, which is in the form YYYY-MM-DD. Here's an extract of my Page model:

class DailyReflectionPage(Page):
    """
    The Daily Reflection Model
    """
    ...
    ...

    reflection_date = models.DateField("Reflection Date", max_length=254)
    ...
    ...
    @cached_property
    def date(self):
        """
        Returns the Reflection's date as a string in %Y-%m-%d format
        """
        fmt = "%Y-%m-%d"
        date_as_string = (self.reflection_date).strftime(fmt)
        return date_as_string      
    ...
    ...
    def full_clean(self, *args, **kwargs):
        # first call the built-in cleanups (including default slug generation)
        super(DailyReflectionPage, self).full_clean(*args, **kwargs)

        # now make your additional modifications
        if self.slug is not self.date:
            self.slug = self.date
    ...
    ...

These daily reflections are written by different authors, as part of a booklet that is published towards the end of the year, for use in the coming year. I would like to have a workflow where, for instance, the daily reflections from January to June are reviewed by one group, and those from July to December are reviewed by another group, as illustrated in the diagram below:

How can this be achieved?

Answer 1

This should be able to be achieved by creating ONE new Workflow Task type that has a relationship to two sets of User Groups (e.g. a/b or before/after, it is probably best to keep this generic in the model definition).

This new Task can be created as part of a new Workflow within the Wagtail admin, and each of the groups linked to the Moderator Group 1 / 2.

Wagtail's methods on the Task allow you to return approval options based on the Page model for any created workflow, from here you can look for a method that would be on the class and assign the groups from there.

The benefits of having a bit more of a generic approach is that you could leverage this for any splitting of moderator assignments as part of future Workflow tasks.

Implementation Overview

• 1 - read the Wagatail Docs on how to add a new Task Type and the Task model reference to understand this process.
• 2 - Read through the full implementation in the code of the built in GroupApprovalTask.
• 3 - In the GroupApprovalTask you can see that the methods with overrides all rely on the checking of self.groups but they all get the page passed in as a arg to those methods.
• 4 - Create a new Task that extends the Wagtail Task class and on this model create two ManyToManyField that allow for two sets of user groups being linked (note: you do not have do to this as two fields, you could put a model in the middle but the example below is just the simplest way to get to the gaol).
• 5 - On the DailyReflectionPage model create a method get_approval_group_key which will return maybe a simple Boolean or a 'A' or 'B' based on the business requirements you described above (check the model's date etc)
• 6 - In your custom Task create a method that abstracts the checking of the Page for this method and returns the Tasks' user group. You may want to add some error handling and default values. E.g. get_approval_groups
• 7 - Add a custom method for each of the 'start', 'user_can_access_editor', page_locked_for_user, user_can_lock, user_can_unlock, get_task_states_user_can_moderate methods that calls get_approval_group with the page and returns the values (see the code GroupApprovalTask for what these should do.

Example Code Snippets

models.py

class DailyReflectionPage(Page):
    """
    The Daily Reflection Model
    """
    def get_approval_group_key(self):
        # custom logic here that checks all the date stuff
        if date_is_after_foo:
            return 'A'
        return 'B'    


class SplitGroupApprovalTask(Task):

    ## note: this is the simplest approach, two fields of linked groups, you could further refine this approach as needed.

    groups_a = models.ManyToManyField(
        Group,
        help_text="Pages at this step in a workflow will be moderated or approved by these groups of users",
        related_name="split_task_group_a",
    )
    groups_b = models.ManyToManyField(
        Group,
        help_text="Pages at this step in a workflow will be moderated or approved by these groups of users",
        related_name="split_task_group_b",
    )

    admin_form_fields = Task.admin_form_fields + ["groups_a", "groups_b"]
    admin_form_widgets = {
        "groups_a": forms.CheckboxSelectMultiple,
        "groups_b": forms.CheckboxSelectMultiple,
    }

    def get_approval_groups(self, page):
       """This method gets used by all checks when determining what group to allow/assign this Task to"""
        
        # recommend some checks here, what if `get_approval_group` is not on the Page?
        approval_group = page.specific.get_approval_group_key()

        if (approval_group == 'A'):
            return self.group_a

        return self.group_b

    # each of the following methods will need to be implemented, all checking for the correct groups for the Page when called
    # def start(self, ...etc)
    # def user_can_access_editor(self, ...etc)
    # def page_locked_for_user(self, ...etc)
    # def user_can_lock(self, ...etc)
    # def user_can_unlock(self, ...etc)


    def get_task_states_user_can_moderate(self, user, **kwargs):
        # Note: this has not been tested, however as this method does not get `page` we must find all the tasks allowed indirectly via their TaskState pages

        tasks = TaskState.objects.filter(status=TaskState.STATUS_IN_PROGRESS, task=self.task_ptr)

        filtered_tasks = []
        for task in tasks:
            page = task.select_related('page_revision', 'task', 'page_revision__page')
            groups = self.get_approval_groups(page)
            if groups.filter(id__in=user.groups.all()).exists() or user.is_superuser:
                filtered_tasks.append(task)

        return TaskState.objects.filter(pk__in=[task.pk for task in filtered_tasks])

    def get_actions(self, page, user):
        # essentially a copy of this method on `GroupApprovalTask` but with the ability to have a dynamic 'group' returned.
        approval_groups = self.get_approval_groups(page)

        if approval_groups.filter(id__in=user.groups.all()).exists() or user.is_superuser:
            return [
                ('reject', "Request changes", True),
                ('approve', "Approve", False),
                ('approve', "Approve with comment", True),
            ]

        return super().get_actions(page, user)