Receiving events from celery task

There are probably many ways to achieve your goal, but here is how I would do it.

Inside your long running celery task set the progress using django's caching framework:

from django.core.cache import cache

@app.task()
def long_running_task(self, *args, **kwargs):
    key = "my_task: %s" % self.result.id
    ...
    # do whatever you need to do and set the progress
    # using cache:
    cache.set(key, progress, timeout="whatever works for you")
    ...

Then all you have to do is make a recurring AJAX GET request with that key and retrieve the progress from cache. Something along those lines:

 def task_progress_view(request, *args, **kwargs):
     key = request.GET.get('task_key')
     progress = cache.get(key)
     return HttpResponse(content=json.dumps({'progress': progress}),
                         content_type="application/json; charset=utf-8")

Here is a caveat though, if you are running your server as multiple processes, make sure that you are using something like memcached, because django's native caching will be inconsistent among the processes. Also I probably wouldn't use celery's task_id as a key, but it is sufficient for demonstration purpose.

Take a look at flower - a real-time monitor and web admin for Celery distributed task queue:

• https://github.com/mher/flower#api
• http://flower.readthedocs.org/en/latest/api.html#get--api-tasks

You need it for presentation, right? Flower works with websockets.

For instance - receive task completion events in real-time (taken from official docs):

var ws = new WebSocket('ws://localhost:5555/api/task/events/task-succeeded/');
ws.onmessage = function (event) {
    console.log(event.data);
}

You would likely need to work with tasks ('ws://localhost:5555/api/tasks/').

I hope this helps.

Simplest:

Your tasks and django app already share access one or two data stores - the broker and the results backend (if you're using one that is different to the broker)

You can simply put some data into one or other of these data stores that indicates which item the task is currently processing.

e.g. if using redis simply have a key 'task-currently-processing' and store the data relevant to the item currenlty being processed in there.

You can use something like Swampdragon to reach the user from the Celery instance (you have to be able to reach it from the client thou, take care not to run afoul of CORS thou). It can be latched onto the counter, not the model itself.

lehins' solution looks good if you don't mind your clients repeatedly polling your backend. That may be fine but it gets expensive as the number of clients grows.

Artur Barseghyan's solution is suitable if you only need the task lifecycle events generated by Celery's internal machinery.

Alternatively, you can use Django Channels and WebSockets to push updates to clients in real-time. Setup is pretty straightforward.

1. Add channels to your INSTALLED_APPS and set up a channel layer. E.g., using a Redis backend:

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("redis", 6379)]
        }
    }
}

2. Create an event consumer. This will receive events from Channels and push them via Websockets to the client. For instance:

import json
from asgiref.sync import async_to_sync
from channels.generic.websocket import WebSocketConsumer


class TaskConsumer(WebsocketConsumer):
    def connect(self):
        self.task_id = self.scope['url_route']['kwargs']['task_id'] # your task's identifier
        async_to_sync(self.channel_layer.group_add)(f"tasks-{self.task_id}", self.channel_name)
        self.accept()

    def disconnect(self, code):
        async_to_sync(self.channel_layer.group_discard)(f"tasks-{self.task_id}", self.channel_name)

    def item_processed(self, event):
        item = event['item']
        self.send(text_data=json.dumps(item))

3. Push events from your Celery tasks like this:

from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer

...
async_to_sync(get_channel_layer.group_send)(f"tasks-{task.task_id}", {
    'type': 'item_processed',
    'item': item,
})

You can also write an async consumer and/or invoke group_send asynchronously. In either case you no longer need the async_to_sync wrapper.

4. Add websocket_urlpatterns to your urls.py:

websocket_urlpatterns = [
    path(r'ws/tasks/<task_id>/', TaskConsumer.as_asgi()),
]

5. Finally, to consume events from JavaScript in your client, you can do something like this:

let task_id = 123;
let protocol = location.protocol === 'https:' ? 'wss://' : 'ws://';
let socket = new WebSocket(`${protocol}${window.location.host}/ws/tasks/${task_id}/`);

socket.onmessage = function(event) {
    let data = JSON.parse(event.data);
    let item = data.item;
    // do something with the item (e.g., push it into your state container)
}