Правильный способ выполнения задачи асинхронно в Django, через Celery

Question

У меня есть приложение Django, которое имитирует Instagram, при этом пользователи загружают фотографии или мемы, а затем их поклонники получают уведомление об указанной фотографии.

В настоящее время для отправки уведомлений, как только пользователь загружает фотографию, я перебираю каждый вентилятор, который имеет загрузчик, добавляет уведомления в список, а затем bulk_create объекты. Как в:

fans = UserFan.objects.filter(star=user).values_list('fan',flat=True) 
    fan_list = [] 
    for fan in fans: 
     fan_list.append(PhotoObjectSubscription(viewer_id=fan, which_photo=photo, updated_at=time, seen=False, type_of_object='1')) 
    PhotoObjectSubscription.objects.bulk_create(fan_list) 

Простые материалы. Обратите внимание, что у меня также есть supervisord, установленный на VM моего приложения, где я запускаю некоторые элементарные задачи через celery (с redis в качестве брокера сообщений).

Теперь я хочу выполнить задачу bulk_create выше как celery task; асинхронно. Мой код bulk_create живет в том же представлении, которое используется для обработки загрузки фотографий, поэтому я считаю, что выполнение этого асинхронно ускорит этот процесс для пользователя.

Я новичок в celery задач, так что кто-то может указать на то, с помощью иллюстративного примера, как я могу включить вышеупомянутую bulk_create задачу в celery задачи? Я сделал исследование, и вот что я думаю, что нужно сделать:

1) Добавить delay() в конце bulk_create заявления:

PhotoObjectSubscription.objects.bulk_create(fan_list).delay() 

2) В tasks.py добавить новую задачу для обработки выше:

@task 
def bulk_create_notifications(): 
    PhotoObjectSubscription.objects.bulk_create(fan_list) 

3) нет необходимости добавлять что-либо к CELERYBEAT_SCHEDULE в settings.py, поскольку задача не является периодической задачей.

Возможно, я не совсем прав, поэтому, пожалуйста, помогите.

Answer 1

Вам необходимо пройти fans (вы, возможно, придется бросить его list, если это на самом деле ValuesListQuerySet), и все другие вещи, потребности задач (например photo.id) в качестве аргумента к задаче:

@task 
def bulk_create_notifications(fans, photo_id): 
    fan_list = [] 
    for fan in fans: 
     fan_list.append(PhotoObjectSubscription(viewer_id=fan, which_photo_id=photo_id, updated_at=time, seen=False, type_of_object='1')) 
    PhotoObjectSubscription.objects.bulk_create(fan_list) 

Затем, вы можете асинхронно запустить задачу с помощью:

# call delay on the task and pass it the same params you would pass to the fnc itself 
bulk_create_notifications.delay(fans) 

Поскольку аргументы должны быть сохранен и передан вашей очередью задач (Redis) можно передавать только аргументы, которые сериализация по сериализатору вас установили в ваших настройках (возможно, JSON). Это означает, что вы должны придерживаться простых типов s.a. строки, целые числа, и вы не можете передавать экземпляры моделей или их списки в качестве параметров.

Вы можете, конечно, начать еще выше и просто пройти user.id и выполнить всю работу db в задаче.