Call Celery class-based task in sync mode

Question

I have celery class-based task

from celery import Task
from django.db import transaction

from config import celery_app


class RefreshData(Task):
    name = "refresh-data"

    @transaction.atomic
    def run(self, *args, **kwargs):
        SomeClass().some_fuc()

celery_app.register_task(RefreshData)

Now I want to run it in sync mode (in tests). How to do that?

Answer 1

The trick is to mock.patch the call of the task in such way, as we call it in sync.

To call task in sync RefreshData().run(*args, **kwargs)

Answer 2

By setting task_always_eager to True, tasks will be executed immediately (synchronously) instead of being sent to the queue (asynchronously), allowing you to debug the code within the task.

task_always_eager is False by default to help prevent inadvertently activating it in production.

reference https://docs.celeryq.dev/en/stable/userguide/configuration.html?highlight=task_always_eager#std-setting-task_always_eager