Как назначить выполнение задачи Celery на поле модели?

Question

[bQ1]

Есть модель Post с полем next_run. Создал тестовые 10 постов. Как запустить задачу celery в время указанное в next_run каждого из объектов модели?

Answer 1

[Сергей Горностаев]

some_task.apply_async(eta=obj.next_run)

Comments

[bQ1]

Спасибо большое, Вы во всех тэгах куратор?) На любой вопрос всегда есть Ваш ответ и всегда в точку)

[Сергей Горностаев]

bQ1, нет, только в тех 53-х, которые хорошо знаю :)

[bQ1]

Сергей Горностаев, не подскажите как запустить сервер redis?
Выполнил pip install redis
Добавил настройки в settings.py
Но не запускается по команде redis-server? (bash: redis-server: command not found)

[Сергей Горностаев]

bQ1, pip install redis устанавливает библиотеку для доступа к Redis из Python, сам Redis надо устанавливать отдельно.

[bQ1]

Сергей Горностаев, он устанавливается глобально или тоже в виртуальное?

[Сергей Горностаев]

bQ1, вам виднее, как вы приложения устанавливаете.

[bQ1]

Сергей Горностаев, в чем проблема:
в settings.py добавил:
CELERY_BROKER_URL = 'redis://localhost:6379'
CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_SERIALIZER = 'json'

Но при запуске celery worker выдает

[2019-08-27 15:28:30,847: ERROR/MainProcess] consumer: Cannot connect to amqp://guest:**@127.0.0.1:5672//: [Errno 61] Connection refused.
Trying again in 2.00 seconds...

Редис сервер запущен
Делал по этому гайду: тут

[Сергей Горностаев]

bQ1, судя по тому, что в ошибке совсем другой URL, вы что-то не так настроили.

[bQ1]

Сергей Горностаев, насчет основного вопроса: например у меня есть 10 объектов с разными датами т/е у меня будет создано 10 задач на эти даты в celery?

[Сергей Горностаев]

bQ1, да.

[bQ1]

Сергей Горностаев, а если допустим пользователи насоздают 10 000 постов, то 10 000 задач это будет нормально?

[Сергей Горностаев]

bQ1, у Celery есть пул процессов, которые забирают задачи из очереди по мере освобождения. Если задач будет много и выполняться они будут долго, то более поздние начнут отставать по срокам выполнения. Стратегию можно поменять, но тогда возникает угроза исчерпания ресурсов сервера.

[bQ1]

Сергей Горностаев, ну, по идее, там задача которая просто меняет дату в поле она не должна сильно задерживать очередь

[Сергей Горностаев]

bQ1, я тоже так думаю. Я думаю, что можно даже извернутся и сворачивать все задачи запланированные на одно время в одну и выполнять изменения в БД одним batch-запросом.

[bQ1]

Сергей Горностаев, а как я могу создавать таск при создании объекта? что нужно писать в save? (если точнее то в какую сторону гуглить? )

[Сергей Горностаев]

bQ1, save не надо трогать. Создайте сигнал post_save и планируйте выполнение задачи в нём.

[bQ1]

Сергей Горностаев, т/е мне нужно в файле tasks.py создать какой то шаблон задачи (например, run), затем на сигнал сделать так:

@receiver(post_save, sender=Posts)
def create_task(sender, instance, **kwargs):
    run.apply_async(eta=instance.next_run)

[Сергей Горностаев]

bQ1, да, всё верно.

[bQ1]

Сергей Горностаев, а сигнал post_save не срабатывает при .update?

[Сергей Горностаев]

bQ1, срабатывает. Можно проверять аргумент created, если нужно реагировать только на создание или только на обновление.

[bQ1]

Сергей Горностаев, задача ставится, все работает но:
из-за чего может не срабатывать повторно post_save?
Сигнал

@receiver(post_save, sender=Posts)
def create_task(sender, instance, **kwargs):
    if instance.autoup:
        interval = instance.autoup_type
        run.apply_async(args=[instance.id, interval], eta=instance.next_run)

Задача:

@task(name="run")
def run(instance_id, interval):
    date = now()
    next_run = date + timedelta(minutes=interval)
    models.Posts.objects.filter(
        pk=instance_id).update(date=date, next_run=next_run)

Суть в том чтобы когда отработает задача, поставилась новая на следующую дату с интервалом interval.
next_run меняется на поставленный интервал но новая задача не ставится

UPD: кажется я сглупил и пишу зря лишний код, когда можно сразу задать периодичность в apply_async?, но все равно не очень понятно почему не работает второй раз