Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Wagtail branching workflows based on value of specified Page field
# see https://stackoverflow.com/questions/69028083/branching-workflows-based-on-value-of-specified-page-field/69060134
# Django imports
from django import forms
from django.db import models
from django.utils.functional import cached_property
# Additional dependencies
# ...
# Wagtail imports
from wagtail.core.models import Group, Page, Task, TaskState, WorkflowState
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
# ...
# ...
@cached_property
def date_in_first_semester(self):
"""
Returns True if Reflection's date
is in the first half of the year
"""
month = (self.reflection_date).month
return month <= 6
def get_approval_group_key(self):
# custom logic here that checks all the date stuff
if self.date_in_first_semester:
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,
verbose_name="for Jan - June Daily Reflections",
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,
verbose_name="for Jul - Dec Daily Reflections",
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?
# here's a simple check
if hasattr(page.specific, "get_approval_group_key"):
approval_group = page.specific.get_approval_group_key()
else:
# arbitrarily assign to group A
# (you could instead do something else)
approval_group = "A"
if approval_group == "A":
return self.groups_a
return self.groups_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, ...etc)
def start(self, workflow_state, user=None):
# essentially a copy of this method on `GroupApprovalTask` but with the ability to have a dynamic 'group' returned.
approval_groups = self.get_approval_groups(workflow_state.page)
if workflow_state.page.locked_by:
# If the person who locked the page isn't in one of the groups, unlock the page
# if not workflow_state.page.locked_by.groups.filter(id__in=self.groups.all()).exists():
if not approval_groups.filter(id__in=self.groups.all()).exists():
workflow_state.page.locked = False
workflow_state.page.locked_by = None
workflow_state.page.locked_at = None
workflow_state.page.save(
update_fields=["locked", "locked_by", "locked_at"]
)
return super().start(workflow_state, user=user)
def user_can_access_editor(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)
# return self.groups.filter(id__in=user.groups.all()).exists() or user.is_superuser
return (
approval_groups.filter(id__in=user.groups.all()).exists()
or user.is_superuser
)
def page_locked_for_user(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)
# return not (self.groups.filter(id__in=user.groups.all()).exists() or user.is_superuser)
return not (
approval_groups.filter(id__in=user.groups.all()).exists()
or user.is_superuser
)
def user_can_lock(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)
# return self.groups.filter(id__in=user.groups.all()).exists()
return approval_groups.filter(id__in=user.groups.all()).exists()
def user_can_unlock(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)
return False
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)
def get_task_states_user_can_moderate(self, user, **kwargs):
# not a very DRY approach, but it works!
if user.is_superuser:
return TaskState.objects.filter(
status=TaskState.STATUS_IN_PROGRESS, task=self.task_ptr
)
elif self.groups_a.filter(id__in=user.groups.all()).exists():
return TaskState.objects.filter(
status=TaskState.STATUS_IN_PROGRESS,
task=self.task_ptr,
workflow_state__in=WorkflowState.objects.filter(
page__in=DailyReflectionPage.objects.filter(
reflection_date__month__lte=6
)
),
)
elif self.groups_b.filter(id__in=user.groups.all()).exists():
return TaskState.objects.filter(
status=TaskState.STATUS_IN_PROGRESS,
task=self.task_ptr,
workflow_state__in=WorkflowState.objects.filter(
page__in=DailyReflectionPage.objects.filter(reflection_date__month__gt=6)
),
)
else:
return TaskState.objects.none()
@classmethod
def get_description(cls):
return "Groups are assigned to approve this task based on reflection month"
class Meta:
verbose_name = "reflection-month-dependent approval task"
verbose_name_plural = "reflection-month-dependent approval tasks"
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment