Add support for GroupResult (#161)

* add group database methods

* remove typo

* encode group results

* use var for using, defaults to None

* use task model in db manager

* try using the _db

* db is not prefixed by _

* remove celery.five usage, split out GroupResult into a unique model

* use to dict for restore group, use different encoding for saving group

* add migration for group result

* add group result admin

* its group_id not task_id

* dont double encode the result

* try using id instead of result

* cleanup group model in cleanup method

* decode result into array of async results

* its to_dict not as_dict

* fix decoding of group result results

* remove unused import

* use autogenerated migration for group result model

* add group views, handle edge cases of group functions

* add tests for group model and views

* undo formatting changes

* fix redefined var and unused import

* resolve multiple leaf in migration graph error

* switch to using path for group result views

* respect settings for id length

* dont add app to GroupResult model creation

* use explicit path to celery content to avoid clobering

* celery imports are weird...need to import states from celery?

* fix flake8 formatting errors

* fix pydocsstyle formatting errors

* appease django by matching the help text exactly

* use custom path converter for task pattern in all paths

* appease flake8 & pydocstyle

* even more flake8 & pydocstyle appeasement

* too many blank lines formatting fix

* use django.urls instead of django.conf.urls

* switch from importing all of celery, to just the stuff we need with renames

* add settings app in group result

* switch to setting up indexes using model meta class

* merge 2 migrations into 1 clean new one

* remove the regex string marker

* fix flake8 formatting complaints on new migration

* revert rogue formatting change

* update introduction docs

Author: AlecTroemel

django_celery_results/admin.py

@@ -11,7 +11,7 @@
     ALLOW_EDITS = False
     pass
 
-from .models import TaskResult
+from .models import TaskResult, GroupResult
 
 
 class TaskResultAdmin(admin.ModelAdmin):
@@ -65,3 +65,17 @@ def get_readonly_fields(self, request, obj=None):
 
 
 admin.site.register(TaskResult, TaskResultAdmin)
+
+
+class GroupResultAdmin(admin.ModelAdmin):
+    """Admin-interface for results  of grouped tasks."""
+
+    model = GroupResult
+    date_hierarchy = 'date_done'
+    list_display = ('group_id', 'date_done')
+    list_filter = ('date_done',)
+    readonly_fields = ('date_created', 'date_done', 'result')
+    search_fields = ('group_id',)
+
+
+admin.site.register(GroupResult, GroupResultAdmin)

django_celery_results/backends/database.py

@@ -10,7 +10,7 @@
 from kombu.exceptions import DecodeError
 from django.db import transaction
 
-from ..models import TaskResult, ChordCounter
+from ..models import TaskResult, ChordCounter, GroupResult
 
 
 logger = get_logger(__name__)
@@ -20,16 +20,23 @@ class DatabaseBackend(BaseDictBackend):
     """The Django database backend, using models to store task state."""
 
     TaskModel = TaskResult
-
+    GroupModel = GroupResult
     subpolling_interval = 0.5
 
-    def _store_result(self, task_id, result, status,
-                      traceback=None, request=None, using=None):
+    def _store_result(
+            self,
+            task_id,
+            result,
+            status,
+            traceback=None,
+            request=None,
+            using=None
+    ):
         """Store return value and status of an executed task."""
         content_type, content_encoding, result = self.encode_content(result)
-        _, _, meta = self.encode_content({
-            'children': self.current_task_children(request),
-        })
+        _, _, meta = self.encode_content(
+            {'children': self.current_task_children(request)}
+        )
 
         task_name = getattr(request, 'task', None)
         worker = getattr(request, 'hostname', None)
@@ -57,8 +64,11 @@ def _store_result(self, task_id, result, status,
             _, _, task_kwargs = self.encode_content(task_kwargs)
 
         self.TaskModel._default_manager.store_result(
-            content_type, content_encoding,
-            task_id, result, status,
+            content_type,
+            content_encoding,
+            task_id,
+            result,
+            status,
             traceback=traceback,
             meta=meta,
             task_name=task_name,
@@ -87,7 +97,9 @@ def _get_task_meta_for(self, task_id):
         # the right names are args/kwargs, not task_args/task_kwargs,
         # keep both for backward compatibility
         res.update(
-            meta, result=result, task_args=task_args,
+            meta,
+            result=result,
+            task_args=task_args,
             task_kwargs=task_kwargs,
             args=task_args,
             kwargs=task_kwargs,
@@ -115,6 +127,36 @@ def _forget(self, task_id):
     def cleanup(self):
         """Delete expired metadata."""
         self.TaskModel._default_manager.delete_expired(self.expires)
+        self.GroupModel._default_manager.delete_expired(self.expires)
+
+    def _restore_group(self, group_id):
+        """return result value for a group by id."""
+        group_result = self.GroupModel._default_manager.get_group(group_id)
+
+        if group_result:
+            res = group_result.as_dict()
+            decoded_result = self.decode_content(group_result, res["result"])
+            res["result"] = (
+                [self.app.AsyncResult(tid) for tid in decoded_result]
+                if decoded_result else None
+            )
+            return res
+
+    def _save_group(self, group_id, group_result):
+        """Store return value of group"""
+        content_type, content_encoding, result = self.encode_content(
+            [r.id for r in group_result]
+        )
+        self.GroupModel._default_manager.store_group_result(
+            content_type, content_encoding, group_id, result
+        )
+        return group_result
+
+    def _delete_group(self, group_id):
+        try:
+            self.GroupModel._default_manager.get_group(group_id).delete()
+        except self.TaskModel.DoesNotExist:
+            pass
 
     def apply_chord(self, header_result_args, body, **kwargs):
         """Add a ChordCounter with the expected number of results"""
@@ -170,14 +212,12 @@ def on_chord_part_return(self, request, state, result, **kwargs):
 
 def trigger_callback(app, callback, group_result):
     """Add the callback to the queue or mark the callback as failed
-
     Implementation borrowed from `celery.app.builtins.unlock_chord`
     """
-    j = (
-        group_result.join_native
-        if group_result.supports_native_join
-        else group_result.join
-    )
+    if group_result.supports_native_join:
+        j = group_result.join_native
+    else:
+        j = group_result.join
 
     try:
         with allow_join_result():

django_celery_results/managers.py

@@ -57,7 +57,44 @@ def _inner(*args, **kwargs):
     return _outer
 
 
-class TaskResultManager(models.Manager):
+class ResultManager(models.Manager):
+    """Generic manager for celery results."""
+
+    def warn_if_repeatable_read(self):
+        if 'mysql' in self.current_engine().lower():
+            cursor = self.connection_for_read().cursor()
+            # MariaDB and MySQL since 8.0 have different transaction isolation
+            # variables: the former has tx_isolation, while the latter has
+            # transaction_isolation
+            if cursor.execute("SHOW VARIABLES WHERE variable_name IN "
+                              "('tx_isolation', 'transaction_isolation');"):
+                isolation = cursor.fetchone()[1]
+                if isolation == 'REPEATABLE-READ':
+                    warnings.warn(TxIsolationWarning(W_ISOLATION_REP.strip()))
+
+    def connection_for_write(self):
+        return connections[router.db_for_write(self.model)]
+
+    def connection_for_read(self):
+        return connections[self.db]
+
+    def current_engine(self):
+        try:
+            return settings.DATABASES[self.db]['ENGINE']
+        except AttributeError:
+            return settings.DATABASE_ENGINE
+
+    def get_all_expired(self, expires):
+        """Get all expired results."""
+        return self.filter(date_done__lt=now() - maybe_timedelta(expires))
+
+    def delete_expired(self, expires):
+        """Delete all expired results."""
+        with transaction.atomic():
+            self.get_all_expired(expires).delete()
+
+
+class TaskResultManager(ResultManager):
     """Manager for :class:`celery.models.TaskResult` models."""
 
     _last_id = None
@@ -136,35 +173,47 @@ def store_result(self, content_type, content_encoding,
             obj.save(using=using)
         return obj
 
-    def warn_if_repeatable_read(self):
-        if 'mysql' in self.current_engine().lower():
-            cursor = self.connection_for_read().cursor()
-            # MariaDB and MySQL since 8.0 have different transaction isolation
-            # variables: the former has tx_isolation, while the latter has
-            # transaction_isolation
-            if cursor.execute("SHOW VARIABLES WHERE variable_name IN "
-                              "('tx_isolation', 'transaction_isolation');"):
-                isolation = cursor.fetchone()[1]
-                if isolation == 'REPEATABLE-READ':
-                    warnings.warn(TxIsolationWarning(W_ISOLATION_REP.strip()))
 
-    def connection_for_write(self):
-        return connections[router.db_for_write(self.model)]
+class GroupResultManager(ResultManager):
+    """Manager for :class:`celery.models.GroupResult` models."""
 
-    def connection_for_read(self):
-        return connections[self.db]
+    _last_id = None
 
-    def current_engine(self):
+    def get_group(self, group_id):
+        """Get result for group by ``group_id``.
+
+        Keyword Arguments:
+        -----------------
+            exception_retry_count (int): How many times to retry by
+                transaction rollback on exception.  This could
+                happen in a race condition if another worker is trying to
+                create the same task.  The default is to retry once.
+
+        """
         try:
-            return settings.DATABASES[self.db]['ENGINE']
-        except AttributeError:
-            return settings.DATABASE_ENGINE
+            return self.get(group_id=group_id)
+        except self.model.DoesNotExist:
+            if self._last_id == group_id:
+                self.warn_if_repeatable_read()
+            self._last_id = group_id
+            return self.model(group_id=group_id)
 
-    def get_all_expired(self, expires):
-        """Get all expired task results."""
-        return self.filter(date_done__lt=now() - maybe_timedelta(expires))
+    @transaction_retry(max_retries=2)
+    def store_group_result(self, content_type, content_encoding,
+                           group_id, result, using=None):
+        fields = {
+            'result': result,
+            'content_encoding': content_encoding,
+            'content_type': content_type,
+        }
 
-    def delete_expired(self, expires):
-        """Delete all expired results."""
-        with transaction.atomic():
-            self.get_all_expired(expires).delete()
+        if not using:
+            using = self.db
+
+        obj, created = self.using(using).get_or_create(group_id=group_id,
+                                                       defaults=fields)
+        if not created:
+            for k, v in fields.items():
+                setattr(obj, k, v)
+            obj.save(using=self.db)
+        return obj