Ref #19 -- Add create_task documentation

Author: codingjoe

docs/inbox.rst

@@ -0,0 +1,44 @@
+=====
+Inbox
+=====
+
+In a multi user workflow it can become handy to assign and reassign tasks and to have
+some kind of inbox to keep track of incomplete tasks.
+
+Automatic assigment
+-------------------
+
+To implement default assignees you need to override implement your own
+:meth:`create_task` method.
+
+
+.. code-block:: python
+
+    from joeflow import tasks
+
+
+    class UpdateWithPrevUserView(tasks.UpdateView):
+
+        def create_task(self, workflow, prev_task):
+            """Assign a new task to the user who completed the previous task."""
+            new_task = workflow.task_set.create(
+                workflow=workflow,
+                name=self.name,
+                type=self.type,
+            )
+            if prev_task.completed_by_user:
+                # If the previous tasks was a machine task, no user is set.
+                new_task.assignees.add(prev_task.completed_by_user)
+            return new_task
+
+
+In this implementation the new task is assigned to the user who completed the previous
+task. The current workflow and the previous task are always given :meth:`create_task`
+method. Note, that a request is not available, since this method might not be called
+within a view but in a machine task. The parent task relation must not be created as a
+part of the :meth:`create_task` method.
+
+API
+===
+
+.. automethod:: joeflow.views.TaskViewMixin.create_task

joeflow/models.py

@@ -571,7 +571,7 @@ def start_next_tasks(self, next_nodes: list = None):
         for node in next_nodes:
             try:
                 # Some nodes – like Join – implement their own method to create new tasks.
-                task = node.create_task(self.workflow)
+                task = node.create_task(self.workflow, self)
             except AttributeError:
                 task = self.workflow.task_set.create(
                     name=node.name, type=node.type, workflow=self.workflow

joeflow/tasks/machine.py

@@ -108,7 +108,7 @@ def __init__(self, *parents: Iterable[str]):
     def __call__(self, workflow, task):
         return set(task.parent_task_set.values_list("name", flat=True)) == self.parents
 
-    def create_task(self, workflow):
+    def create_task(self, workflow, prev_task):
         return workflow.task_set.get_or_create(
             name=self.name,
             type=self.type,

joeflow/views.py

@@ -71,6 +71,22 @@ def post(self, request, *args, **kwargs):
         task.start_next_tasks()
         return response
 
+    def create_task(self, workflow, prev_task):
+        """
+        Factory method retuning a new task instance.
+
+        Args:
+            workflow (joeflow.models.Workflow): Current workflow instance.
+            prev_task (joeflow.models.Task): Instance of the previous Task.
+
+        Returns:
+            joeflow.models.Task: New task instance.
+
+        """
+        return workflow.task_set.create(
+            name=self.name, type=self.type, workflow=workflow
+        )
+
 
 class WorkflowDetailView(WorkflowTemplateNameViewMixin, generic.DetailView):
     pass

tests/test_views.py

@@ -0,0 +1,30 @@
+from django.urls import reverse
+
+from tests.testapp import workflows
+
+
+class TestTaskViewMixin:
+
+    def test_create_task(self, db, admin_client, admin_user):
+        url = reverse('simpleworkflow:start_view')
+        response = admin_client.post(url)
+        assert response.status_code == 302
+
+        wf = workflows.SimpleWorkflow.objects.get()
+
+        assert wf.task_set.count() == 2
+        new_task = wf.task_set.get(name='save_the_princess')
+        assert new_task.name == 'save_the_princess'
+        assert new_task.type == 'human'
+        assert not new_task.assignees.all()
+
+    def test_custom_create_task(self, db, admin_client, admin_user):
+        url = reverse('assigneeworkflow:start_view')
+        response = admin_client.post(url)
+        assert response.status_code == 302
+
+        wf = workflows.AssigneeWorkflow.objects.get()
+
+        assert wf.task_set.count() == 2
+        new_task = wf.task_set.get(name='save_the_princess')
+        assert admin_user in new_task.assignees.all()

tests/testapp/urls.py

@@ -22,6 +22,7 @@
     path("admin/", admin.site.urls),
     path("shipment/", include(workflows.ShippingWorkflow.urls())),
     path("simple/", include(workflows.SimpleWorkflow.urls())),
+    path("assignee/", include(workflows.AssigneeWorkflow.urls())),
     path("gateway/", include(workflows.GatewayWorkflow.urls())),
     path("splitjoin/", include(workflows.SplitJoinWorkflow.urls())),
     path("loop/", include(workflows.LoopWorkflow.urls())),

tests/testapp/views.py

@@ -0,0 +1,16 @@
+from joeflow import tasks
+
+
+class UpdateWithPrevUserView(tasks.UpdateView):
+
+    def create_task(self, workflow, prev_task):
+        """Assign a new task to the user who completed the previous task."""
+        new_task = workflow.task_set.create(
+            workflow=workflow,
+            name=self.name,
+            type=self.type,
+        )
+        if prev_task.completed_by_user:
+            # If the previous tasks was a machine task, no user is set.
+            new_task.assignees.add(prev_task.completed_by_user)
+        return new_task

tests/testapp/workflows.py

@@ -5,6 +5,7 @@
 from joeflow import tasks
 
 from . import models
+from .views import UpdateWithPrevUserView
 
 
 class ShippingWorkflow(models.Shipment):
@@ -86,6 +87,24 @@ class Meta:
         proxy = True
 
 
+class AssigneeWorkflow(models.SimpleWorkflowState):
+    start_view = tasks.StartView(fields="__all__", path="custom/postfix/")
+    start_method = tasks.Start()
+    save_the_princess = UpdateWithPrevUserView(fields="__all__")
+
+    def end(self):
+        pass
+
+    edges = (
+        (start_view, save_the_princess),
+        (start_method, save_the_princess),
+        (save_the_princess, end),
+    )
+
+    class Meta:
+        proxy = True
+
+
 class GatewayWorkflow(models.GatewayWorkflowState):
     start = tasks.StartView(fields="__all__")