Должна ли задача сельдерея немедленно сохранять обновления модели django?

Question

У меня есть задача celery, которая выполняется по пять минут за раз, чтобы проверить тему подписки Azure Service Bus на наличие сообщений. В течение пяти минут он проверяет сервисную шину каждую секунду, чтобы проверить наличие сообщения, и, если он его находит, сохраняет некоторую информацию в базе данных. Я заметил, что фактическая фиксация в базе данных происходит только тогда, когда задача заканчивается на пятиминутной отметке, а не когда вызывается метод model.save ().

Мне интересно, стоит ли добавлять код для принудительного выполнения каждого сохранения немедленно, а не в конце пяти минут? Я думаю об использовании оператора, который включает в себя атомарные транзакции для достижения этой цели.

Код ниже содержит мою задачу. Я использую цикл while, чтобы задание продолжалось в течение 5 минут, и внутри него я пингую служебную шину для сообщений каждую секунду, а затем сохраняю их, если они найдены.

class CheckForUpdates(PeriodicTask):
    run_every = 300

    def run(self, queue_name='bus_queue'):
        end_task_time = _at_five_minutes()
        while time.time() < end_task_time:
            _wait_for_one_second()
            result = _check_service_bus_for_update()
            if _update_was_found(result):
                update = json.loads(result.body)
                logger.info("azure response body: ", update)
                # code that updates a django model
                model.save()

Это хороший дизайн? Можно ли накапливать фиксации базы данных в течение 5 минут, а затем сохранять их все последовательно в конце 5 минут? Стоит ли использовать транзакции или принудительно сохранять задачу каждый раз?

Answer 1

Я рекомендую использовать сигнал моделей в Django, чтобы вы могли ждать прослушивания сигнала из баз данных. django.db.models.signals имел некоторый хук, который вы можете использовать, так что вы можете просто ждать, когда произойдет событие, я буду опубликовать пример:

signal.py
from django.db.models.signals import post_save
from pay.models import RateDeck

@receiver(post_save, sender=RateDeck)
def post_save_RateDeck(sender, instance, **kwargs):
idpk = instance.pk
transaction.on_commit(lambda :uploadrate.apply_async(args=[idpk]))

Поэтому я надеюсь, что вы получите, что uploadrate - это функция, которая вызывается, когда транзакция завершена. transaction.on_commit отправляет сигнал, когда транзакция в базах данных завершена.