celery_task_socket_io.py

import celery as celery

from extensions import socketio


@celery.task(name="some_task", bind=True)
def some_task(self, room_name=None, namespace=None):
    # Do some task
    # Send a event to room
    socketio.emit(
        "some_event", dict(random_key="random_value"),
        namespace=namespace, room=room,
        include_self=False, skip_sid=["testing"]
    )
    return "succeeded"