Update model structure

Author: codingjoe

galahad/admin.py

@@ -14,7 +14,7 @@ def rerun(modeladmin, request, queryset):
         if succeeded:
             messages.warning(request, "Only failed tasks can be retried. %s tasks have been skipped" % succeeded)
         counter = 0
-        for obj in queryset.filter(~queryset.is_completed).iterator():
+        for obj in queryset.not_succeeded().iterator():
             obj.enqueue()
             counter += 1
         messages.success(request, "%s tasks have been successfully queued" % counter)
@@ -37,32 +37,48 @@ def child_tasks(self, obj):
     actions = ('rerun',)
 
     list_display = (
-        'content_type',
         'node_name',
+        'status',
+        'node_type',
+        'content_type',
         'completed',
-        'failed',
+        'modified',
+        'created',
     )
 
     readonly_fields = (
         'process',
         'node_name',
+        'node_type',
         'parent_task_set',
         'child_tasks',
         'completed',
-        'failed',
+        'created',
+        'modified',
         'exception',
         'pretty_stacktrace',
     )
 
     list_filter = (
-        'completed',
-        'failed',
+        'status',
+        'node_type',
         'content_type',
+        'completed',
+        'created',
     )
 
     fieldsets = (
         (None, {
-            'fields': ('process', 'node_name', 'parent_task_set', 'child_tasks', 'completed', 'failed', 'exception')
+            'fields': (
+                'process',
+                'node_name',
+                'parent_task_set',
+                'child_tasks',
+                'completed',
+                'modified',
+                'created',
+                'exception'
+            ),
         }),
         (t('Traceback'), {
             'classes': ('collapse',),

galahad/migrations/0001_initial.py

@@ -1,5 +1,4 @@
 import logging
-import random
 import sys
 import traceback
 
@@ -70,8 +69,8 @@ class Process(models.Model, metaclass=BaseProcess):
     :attr:`.edges`.
     """
     id = models.BigAutoField(primary_key=True, editable=False)
-    started = models.DateTimeField(auto_now_add=True, db_index=True)
-    completed = models.DateTimeField(blank=True, null=True, editable=False, db_index=True)
+    created = models.DateTimeField(auto_now_add=True, db_index=True)
+    modified = models.DateTimeField(auto_now=True, db_index=True)
 
     task_set = GenericRelation('galahad.Task', object_id_field='_process_id')
 
@@ -267,60 +266,112 @@ def get_instance_graph_svg(self, output_format='svg'):
         graph.format = output_format
         return SafeString(graph.pipe().decode('utf-8'))
 
+    def save(self, **kwargs):
+        if self.pk:
+            try:
+                update_fields = kwargs['update_fields']
+            except KeyError:
+                pass
+            else:
+                update_fields.append('modified')
+        super().save(**kwargs)
+
 
 def process_subclasses():
     from django.apps import apps
 
     apps.check_models_ready()
     query = models.Q()
-    for app_models in apps.all_models.values():
-        for model in app_models.values():
-            if issubclass(model, Process) and model is not Process:
-                opts = model._meta
-                query |= models.Q(app_label=opts.app_label, model=opts.model_name)
+    for model in apps.get_models():
+        if issubclass(model, Process) and model is not Process:
+            opts = model._meta
+            query |= models.Q(app_label=opts.app_label, model=opts.model_name)
     return query
 
 
 class TasksQuerySet(models.query.QuerySet):
-    is_completed = models.Q(completed__isnull=False)
-    is_failed = models.Q(failed__isnull=False)
 
     def scheduled(self):
-        return self.filter(~(self.is_failed | self.is_completed))
+        return self.filter(status=self.model.SCHEDULED)
 
     def succeeded(self):
-        return self.filter(self.is_completed)
+        return self.filter(status=self.model.SUCCEEDED)
+
+    def not_succeeded(self):
+        return self.exclude(status=self.model.SUCCEEDED)
 
     def failed(self):
-        return self.filter(self.is_failed)
+        return self.filter(status=self.model.FAILED)
 
 
 class Task(models.Model):
     id = models.BigAutoField(primary_key=True, editable=False)
-    _process = models.ForeignKey('galahad.Process', on_delete=models.CASCADE, db_column='process_id', editable=False)
+    _process = models.ForeignKey(
+        'galahad.Process',
+        on_delete=models.CASCADE,
+        db_column='process_id'
+        , editable=False,
+    )
     content_type = models.ForeignKey(
-        'contenttypes.ContentType', on_delete=models.CASCADE,
-        editable=False, limit_choices_to=process_subclasses,
+        'contenttypes.ContentType',
+        on_delete=models.CASCADE,
+        editable=False,
+        limit_choices_to=process_subclasses,
         related_name='galahad_task_set',
     )
     process = GenericForeignKey('content_type', '_process_id')
+
     node_name = models.TextField(db_index=True, editable=False)
-    parent_task_set = models.ManyToManyField('self', related_name='child_task_set', editable=False, symmetrical=False)
 
-    assignees = models.ManyToManyField(settings.AUTH_USER_MODEL, verbose_name=t('assignees'), related_name='galahad_task_set')
+    HUMAN = 'human'
+    MACHINE = 'machine'
+    _node_type_choices = (
+        (HUMAN, t(HUMAN)),
+        (MACHINE, t(MACHINE)),
+    )
+    node_type = models.TextField(
+        choices=_node_type_choices,
+        editable=False,
+        db_index=True,
+    )
+
+    parent_task_set = models.ManyToManyField(
+        'self',
+        related_name='child_task_set',
+        editable=False,
+        symmetrical=False,
+    )
+
+    FAILED = 'failed'
+    SUCCEEDED = 'succeeded'
+    SCHEDULED = 'scheduled'
+    _status_choices = (
+        (FAILED, t(FAILED)),
+        (SUCCEEDED, t(SUCCEEDED)),
+        (SCHEDULED, t(SCHEDULED)),
+    )
+    status = models.TextField(
+        choices=_status_choices,
+        default=SCHEDULED,
+        editable=False,
+        db_index=True,
+    )
+
+    assignees = models.ManyToManyField(
+        settings.AUTH_USER_MODEL,
+        verbose_name=t('assignees'),
+        related_name='galahad_task_set',
+    )
 
     created = models.DateTimeField(auto_now_add=True, db_index=True)
+    modified = models.DateTimeField(auto_now=True, db_index=True)
     completed = models.DateTimeField(blank=True, null=True, editable=False, db_index=True)
-    failed = models.DateTimeField(blank=True, null=True, editable=False, db_index=True)
 
     exception = models.TextField(blank=True)
     stacktrace = models.TextField(blank=True)
 
     objects = TasksQuerySet.as_manager()
 
-    def __str__(self):
-        return '%s (%s)' % (self.node_name, self.pk)
-
     class Meta:
         ordering = ('-completed', '-created')
         get_latest_by = ('created',)
@@ -330,6 +381,21 @@ class Meta:
         )
         default_manager_name = 'objects'
 
+    def __str__(self):
+        return '%s (%s)' % (self.node_name, self.pk)
+
+    def save(self, **kwargs):
+        if self.pk:
+            try:
+                update_fields = kwargs['update_fields']
+            except KeyError as e:
+                raise ValueError(
+                    "You need to provide explicit 'update_fields' to avoid race conditions."
+                ) from e
+            else:
+                update_fields.append('modified')
+        super().save(**kwargs)
+
     def get_absolute_url(self):
         if self.completed:
             return
@@ -345,17 +411,19 @@ def node(self):
 
     def finish(self):
         self.completed = timezone.now()
+        self.status = self.SUCCEEDED
         if self.pk:
-            self.save(update_fields=['completed'])
+            self.save(update_fields=['status', 'completed'])
         else:
             self.save()
 
     def fail(self):
-        self.failed = timezone.now()
+        self.completed = timezone.now()
+        self.status = self.FAILED
         tb = traceback.format_exception(*sys.exc_info())
         self.exception = tb[-1].strip()
         self.stacktrace = "".join(tb)
-        self.save(update_fields=['failed', 'exception', 'stacktrace'])
+        self.save(update_fields=['status', 'exception', 'stacktrace'])
 
     def enqueue(self, countdown=None, eta=None):
         """
@@ -372,12 +440,22 @@ def enqueue(self, countdown=None, eta=None):
             celery.result.AsyncResult: Celery task result.
 
         """
-        return celery.task_wrapper.apply_async(
+        self.status = self.SCHEDULED
+        self.completed = None
+        self.exception = ''
+        self.stacktrace = ''
+        self.save(update_fields=[
+            'status',
+            'completed',
+            'exception',
+            'stacktrace',
+        ])
+        transaction.on_commit(lambda: celery.task_wrapper.apply_async(
             args=(self.pk, self._process_id),
             countdown=countdown,
             eta=eta,
             queue=settings.GALAHAD_CELERY_QUEUE_NAME,
-        )
+        ))
 
     @transaction.atomic()
     def start_next_tasks(self, next_nodes: list = None):
@@ -400,11 +478,12 @@ def start_next_tasks(self, next_nodes: list = None):
                 # Some nodes – like Join – implement their own method to create new tasks.
                 task = node.create_task(self.process)
             except AttributeError:
-                task = self.process.task_set.create(node_name=node.node_name, completed=None, failed=None)
+                task = self.process.task_set.create(
+                    node_name=node.node_name,
+                    node_type=node.node_type,
+                )
             task.parent_task_set.add(self)
             if callable(node):
                 transaction.on_commit(task.enqueue)
             tasks.append(task)
-        else:
-            self.process.finish()
         return tasks

galahad/migrations/__init__.py

@@ -101,7 +101,11 @@ def __call__(self, process, task):
         return set(task.parent_task_set.values_list('node_name', flat=True)) == self.parents
 
     def create_task(self, process):
-        return process.task_set.get_or_create(node_name=self.node_name, completed=None, failed=None)[0]
+        return process.task_set.get_or_create(
+            node_name=self.node_name,
+            node_type=self.node_type,
+            completed=None,
+        )[0]
 
 
 class NonBlockingJoin(Join):
@@ -111,7 +115,10 @@ def __call__(self, process, task):
             return tuple()
 
     def create_task(self, process):
-        return process.task_set.get_or_create(node_name=self.node_name, failed=None)[0]
+        return process.task_set.get_or_create(
+            node_name=self.node_name,
+            node_type=self.node_type,
+        )[0]
 
 
 class Wait:

galahad/models.py

@@ -30,6 +30,7 @@ def test_create_task(self, db):
         proc = models.SimpleProcess.start_method()
         node = tasks.Join()
         node.node_name = 'test'
+        node.node_type = 'machine'
         obj = node.create_task(proc)
         obj2 = node.create_task(proc)
         assert obj == obj2

galahad/tasks/machine.py

@@ -1,6 +1,7 @@
 """High level integration tests."""
 import sys
 import time
+from unittest.mock import patch
 
 import pytest
 from django.urls import reverse
@@ -61,7 +62,8 @@ def test_start_method(self, db):
 
 class TestSplitJoinProcess:
 
-    def test_join(self, db, client):
+    @patch('galahad.celery.task_wrapper.retry')
+    def test_join(self, retry, db, client):
         url = reverse('splitjoinprocess:start')
         response = client.post(url)
         assert response.status_code == 302