Merge pull request #115 from czue/group-results

support group results

README.md

@@ -149,6 +149,43 @@ $(function () {
 });
 ```
 
+### Working with Groups
+
+This library includes experimental support for working with [Celery groups](https://docs.celeryq.dev/en/stable/userguide/canvas.html#groups).
+You can use the `"group_status"` URL endpoint for this. Here is a basic example:
+
+**Example task:**
+
+```python
+@shared_task(bind=True)
+def add(self, x, y):
+    return x + y
+```
+
+**Calling view:**
+
+```python
+from celery import group
+from .tasks import add
+
+def progress_view(request):
+    task_group = group(add.s(i, i) for i in range(100))
+    group_result = task_group.apply_async()
+    # you must explicitly call the save function on the group_result after calling the tasks
+    group_result.save()
+    return render(request, 'display_progress.html', context={'task_id': group_result.id})
+
+```
+
+**Template:**
+
+```html
+document.addEventListener("DOMContentLoaded", function () {
+  var progressUrl = "{% url 'celery_progress:group_status' task_id %}";
+  CeleryProgressBar.initProgressBar(progressUrl);
+});
+```
+
 ## Customization
 
 The `initProgressBar` function takes an optional object of options. The following options are supported:

celery_progress/backend.py

@@ -4,7 +4,7 @@
 from abc import ABCMeta, abstractmethod
 from decimal import Decimal
 
-from celery.result import EagerResult, allow_join_result
+from celery.result import EagerResult, allow_join_result, AsyncResult
 from celery.backends.base import DisabledBackend
 
 logger = logging.getLogger(__name__)
@@ -192,3 +192,34 @@ def _get_unknown_progress(state):
         'total': 100,
         'percent': 0,
     }
+
+
+class GroupProgress:
+
+    def __init__(self, group_result):
+        """
+        group_result:
+            a GroupResult or an object that mimics it to a degree
+        """
+        self.group_result = group_result
+
+    def get_info(self):
+        if not self.group_result.children:
+            raise Exception("There were no tasks to track in the group!")
+        else:
+            child_progresses = [Progress(child) for child in self.group_result.children]
+            child_infos = [cp.get_info() for cp in child_progresses]
+            child_progress_dicts = [ci["progress"] for ci in child_infos]
+            total = sum(cp["total"] for cp in child_progress_dicts)
+            current = sum(cp["current"] for cp in child_progress_dicts)
+            percent = float(round(100 * current / total, 2))
+            info = {
+                "complete": all(ci["complete"] for ci in child_infos),
+                "success": all(ci["success"] for ci in child_infos),
+                "progress": {
+                    "total": total,
+                    "current": current,
+                    "percent": percent,
+                }
+            }
+            return info

celery_progress/urls.py

@@ -3,5 +3,6 @@
 
 app_name = 'celery_progress'
 urlpatterns = [
-    re_path(r'^(?P<task_id>[\w-]+)/$', views.get_progress, name='task_status')
+    re_path(r'^(?P<task_id>[\w-]+)/$', views.get_progress, name='task_status'),
+    re_path(r'^g/(?P<group_id>[\w-]+)/$', views.get_group_progress, name='group_status')
 ]

celery_progress/views.py

@@ -1,10 +1,17 @@
 import json
 from django.http import HttpResponse
-from celery.result import AsyncResult
-from celery_progress.backend import Progress
+from celery.result import AsyncResult, GroupResult
+from celery_progress.backend import Progress, GroupProgress
 from django.views.decorators.cache import never_cache
 
 @never_cache
 def get_progress(request, task_id):
     progress = Progress(AsyncResult(task_id))
     return HttpResponse(json.dumps(progress.get_info()), content_type='application/json')
+
+
+
+@never_cache
+def get_group_progress(request, group_id):
+    group_progress = GroupProgress(GroupResult.restore(group_id))
+    return HttpResponse(json.dumps(group_progress.get_info()), content_type='application/json')