Skip to content

Commit

Permalink
Factor out workflow item management helper functions into view mixin (#…
Browse files Browse the repository at this point in the history 
…1346)
  • Loading branch information
@dalecannon
dalecannon authored Dec 4, 2024
1 parent 708944c commit 02044ba
Showing 1 changed file with 93 additions and 103 deletions.
196 changes: 93 additions & 103 deletions tasks/views.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@
from tasks.forms import TaskWorkflowTemplateCreateForm
from tasks.forms import TaskWorkflowTemplateDeleteForm
from tasks.forms import TaskWorkflowTemplateUpdateForm
from tasks.models import Queue
from tasks.models import QueueItem
from tasks.models import Task
from tasks.models import TaskItem
from tasks.models import TaskItemTemplate
Expand Down Expand Up @@ -262,25 +264,98 @@ def get_context_data(self, **kwargs):
return context_data


class TaskWorkflowDetailView(PermissionRequiredMixin, DetailView):
model = TaskWorkflow
template_name = "tasks/workflows/detail.jinja"
permission_required = "tasks.view_taskworkflow"
class QueuedItemManagementMixin:
"""A view mixin providing helper functions to manage queued items."""

queued_item_model: type[QueueItem] = None
"""The model responsible for managing members of a queue."""

item_lookup_field: str = ""
"""The lookup field of the instance managed by a queued item."""

queue_field: str = ""
"""The name of the ForeignKey field relating a queued item to a queue."""

@cached_property
def task_workflow(self) -> TaskWorkflow:
def queue(self) -> type[Queue]:
"""The queue instance that is the object of the view."""
return self.get_object()

def promote(self, lookup_id: int) -> None:
queued_item = get_object_or_404(
self.queued_item_model,
**{
self.item_lookup_field: lookup_id,
self.queue_field: self.queue,
},
)
try:
queued_item.promote()
except OperationalError:
pass

def demote(self, lookup_id: int) -> None:
queued_item = get_object_or_404(
self.queued_item_model,
**{
self.item_lookup_field: lookup_id,
self.queue_field: self.queue,
},
)
try:
queued_item.demote()
except OperationalError:
pass

def promote_to_first(self, lookup_id: int) -> None:
queued_item = get_object_or_404(
self.queued_item_model,
**{
self.item_lookup_field: lookup_id,
self.queue_field: self.queue,
},
)
try:
queued_item.promote_to_first()
except OperationalError:
pass

def demote_to_last(self, lookup_id: int) -> None:
queued_item = get_object_or_404(
self.queued_item_model,
**{
self.item_lookup_field: lookup_id,
self.queue_field: self.queue,
},
)
try:
queued_item.demote_to_last()
except OperationalError:
pass


class TaskWorkflowDetailView(
PermissionRequiredMixin,
QueuedItemManagementMixin,
DetailView,
):
template_name = "tasks/workflows/detail.jinja"
permission_required = "tasks.view_taskworkflow"
model = TaskWorkflow
queued_item_model = TaskItem
item_lookup_field = "task_id"
queue_field = "queue"

@property
def view_url(self) -> str:
return reverse(
"workflow:task-workflow-ui-detail",
kwargs={"pk": self.task_workflow.pk},
kwargs={"pk": self.queue.pk},
)

def get_context_data(self, **kwargs):
context_data = super().get_context_data(**kwargs)
context_data["object_list"] = self.task_workflow.get_tasks()
context_data["object_list"] = self.queue.get_tasks()
return context_data

def post(self, request, *args, **kwargs):
Expand All @@ -295,50 +370,6 @@ def post(self, request, *args, **kwargs):

return HttpResponseRedirect(self.view_url)

def promote(self, task_id: int) -> None:
task_item = get_object_or_404(
TaskItem,
task_id=task_id,
queue=self.task_workflow,
)
try:
task_item.promote()
except OperationalError:
pass

def demote(self, task_id: int) -> None:
task_item = get_object_or_404(
TaskItem,
task_id=task_id,
queue=self.task_workflow,
)
try:
task_item.demote()
except OperationalError:
pass

def promote_to_first(self, task_id: int) -> None:
task_item = get_object_or_404(
TaskItem,
task_id=task_id,
queue=self.task_workflow,
)
try:
task_item.promote_to_first()
except OperationalError:
pass

def demote_to_last(self, task_id: int) -> None:
task_item = get_object_or_404(
TaskItem,
task_id=task_id,
queue=self.task_workflow,
)
try:
task_item.demote_to_last()
except OperationalError:
pass


class TaskWorkflowCreateView(PermissionRequiredMixin, FormView):
permission_required = "tasks.add_taskworkflow"
Expand Down Expand Up @@ -427,25 +458,28 @@ def get_context_data(self, **kwargs):
return context_data


class TaskWorkflowTemplateDetailView(PermissionRequiredMixin, DetailView):
model = TaskWorkflowTemplate
class TaskWorkflowTemplateDetailView(
PermissionRequiredMixin,
QueuedItemManagementMixin,
DetailView,
):
template_name = "tasks/workflows/detail.jinja"
permission_required = "tasks.view_taskworkflowtemplate"

@cached_property
def task_workflow_template(self) -> TaskWorkflowTemplate:
return self.get_object()
model = TaskWorkflowTemplate
queued_item_model = TaskItemTemplate
item_lookup_field = "task_template_id"
queue_field = "queue"

@property
def view_url(self) -> str:
return reverse(
"workflow:task-workflow-template-ui-detail",
kwargs={"pk": self.task_workflow_template.pk},
kwargs={"pk": self.queue.pk},
)

def get_context_data(self, **kwargs):
context_data = super().get_context_data(**kwargs)
context_data["object_list"] = self.task_workflow_template.get_task_templates()
context_data["object_list"] = self.queue.get_task_templates()
return context_data

def post(self, request, *args, **kwargs):
Expand All @@ -460,50 +494,6 @@ def post(self, request, *args, **kwargs):

return HttpResponseRedirect(self.view_url)

def promote(self, task_template_id: int) -> None:
task_item_template = get_object_or_404(
TaskItemTemplate,
task_template_id=task_template_id,
queue=self.task_workflow_template,
)
try:
task_item_template.promote()
except OperationalError:
pass

def demote(self, task_template_id: int) -> None:
task_item_template = get_object_or_404(
TaskItemTemplate,
task_template_id=task_template_id,
queue=self.task_workflow_template,
)
try:
task_item_template.demote()
except OperationalError:
pass

def promote_to_first(self, task_template_id: int) -> None:
task_item_template = get_object_or_404(
TaskItemTemplate,
task_template_id=task_template_id,
queue=self.task_workflow_template,
)
try:
task_item_template.promote_to_first()
except OperationalError:
pass

def demote_to_last(self, task_template_id: int) -> None:
task_item_template = get_object_or_404(
TaskItemTemplate,
task_template_id=task_template_id,
queue=self.task_workflow_template,
)
try:
task_item_template.demote_to_last()
except OperationalError:
pass


class TaskWorkflowTemplateCreateView(PermissionRequiredMixin, CreateView):
model = TaskWorkflowTemplate
Expand Down

0 comments on commit 02044ba

Please sign in to comment.