Django + DRF + Celery: execute celery_task.delay() only after database transaction completed (model.save() reached the database)

user_9016861 0
Viewed 36

I tried to use the on_commit method as they tell us in the docs, but I still get ModelDoesNotExistError inside the task launched via delay().

views.py:

class SomeViewset(viewsets.ModelViewset):
    ...
    @action(detail=True, methods=['post'])
    def some_action(self, request, *args, **kwargs):
        m = MyModel()

        with transaction.atomic():
            m.save()
            transaction.on_commit(lambda:
                                      my_fav_task.delay(m.id, param2, param3))
        return Response({"success": True, "operation_id": m.id}, status=status.HTTP_200_OK)

tasks.py:

@shared_task
def my_fav_task(operation_id, **params):
    print(operation_id)  # We get new id
    print(MyModel.objects.all())  # No newly created object here
    operation = MyModel.objects.get(id=operation_id)  # Error here
django-rest-framework django-celery celery atomic django
1 Answers

My problem was in not obvious usage of different databases between celery and django. Seems that docker containers executing this processes created local db each for their own, because I unwillingly used standard db.sqlite3 without special sharing it. After changing DB settings to postgre, which was running in another container, the following pattern worked:

from django.db import transactions

@transaction.atomic
def view(request):
    m = MyModel()
    m.save()
    # Execute celery task only after database changes (m.save())
    transaction.on_commit(lambda: my_fav_task.delay(m.id, param2, param3))
    return Response({"success": True, "operation_id": m.id}, status=status.HTTP_200_OK)
user_90168610
Related