How not to go into infinite recursion with a Django signal and a Celery task

Question

I'm working on billing Django app and should generate Pdf from budget on background task each time any value changes.

models.py:

from django_model_changes import ChangesMixin

class Budget(ChangesMixin, models.Model):
  pdf = models.FileField(verbose_name="PDF", max_length=200, upload_to='pdf/budgets', null=True, blank=True)

tasks.py:

@shared_task(name='generate_budget_pdf_task')
def generate_budget_pdf_task(budget_id):
  budget = get_object_or_404(app_models.Budget, id=id)
  concepts = budget.concept_budget.all()
  try:
    file = app_pdf.PDF(None, budget, concepts)
    pdf_file = file.generate_budget()
    file_name = 'Pto_' + str(budget.brand.client.name) + '_' + str(budget.name) + '_' + str(budget.code) + ".pdf"
    budget.pdf.save(file_name, ContentFile(pdf_file), save=True)

  except Exception as e:
    print("Error generate_budget_pdf_task")
    print(e)
    print(type(e))

signals.py:

def generate_budget_pdf_signal(sender, instance, created, **kwargs):
  if 'pdf' not in instance.changes():
    app_tasks.generate_budget_pdf_task.delay(instance.id)

post_save.connect(generate_budget_pdf_signal, sender=app_models.Budget)

But the task update

Any could help me please ? Thanks in advance.

Answer 1

Here is the code that solves the problem

models.py:

class Budget(ChangesMixin, models.Model):
  def save(self, *args, **kwargs):
    try:
      if self._generated_pdf == False:
        self._generated_pdf = True
    except:
      self._generated_pdf = False

signals.py:

def generate_budget_pdf_signal(sender, instance, created, **kwargs):
  if 'pdf' not in instance.changes():
    app_tasks.generate_budget_pdf_task.delay(instance.id)

post_save.connect(generate_budget_pdf_signal, sender=app_models.Budget)

tasks.py:

@shared_task(name='generate_budget_pdf_task')
def generate_budget_pdf_task(budget_id):
  budget = get_object_or_404(app_models.Budget, id=budget_id)
  concepts = budget.concept_budget.all()
  try:
    file = app_pdf.PDF(None, budget, concepts)
    pdf_file = file.generate_budget()
    file_name = 'Pto_' + str(budget.brand.client.name) + '_' + str(budget.name) + '_' + str(budget.code) + ".pdf"
    budget._generated_pdf = True
    budget.pdf.save(file_name, ContentFile(pdf_file), save=True)
  except Exception as e:
    print("Error generate_budget_pdf_task")
    print(e)
    print(type(e))