custom_celery_task.py

"""
Custom Celery task for providing useful extensions
http://docs.celeryproject.org/en/latest/userguide/tasks.html#custom-task-classes
"""
from celery.app.task import Task
from django.db import transaction


class BetterCeleryTask(Task):
    """Extend the default Task class to add helpers with Django's on_commit()."""

    def delay_on_commit(self, *args, **kwargs):
        return transaction.on_commit(lambda: self.delay(*args, **kwargs))

    def apply_async_on_commit(self, *args, **kwargs):
        return transaction.on_commit(lambda: self.apply_async(*args, **kwargs))


# use it per task as follows:
@app.task(base=BetterCeleryTask)
def some_task(some_item_pk):
    item = Item.objects.get(pk=some_item_pk)
    print(item.name)


# or for the whole app, in which case all your tasks will use it
celery_app = celery.Celery(
    "your_app_name", 
    task_cls="your.python.path:BetterCeleryTask",
)


# And use your taks as follows
some_task.delay_on_commit(123)