Создание PeriodicTask в celery django

Мне нужно отправлять уведомления с определенным интервалом, когда экземпляр модели Ticket был создан. Для этого я решил использовать celery-beat.

Я создал signals.py, где создаю инстанс PeriodicTask. Когда я создаю новый экземпляр Ticket, экземпляр PeriodicTask создается в DB, но задача не запускается. Что я делаю не так?

signals.py

from datetime import datetime
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Ticket
from django_celery_beat.models import PeriodicTask, IntervalSchedule
import json

@receiver(post_save,sender=Ticket)
def notification_handler(sender, instance, created, **kwargs):
    if created:
        interval, created = IntervalSchedule.objects.get_or_create(every=10, period='seconds')
        task = PeriodicTask.objects.create(interval=interval,enabled=True, name='notification_' + str(instance.id), task="create", args=json.dumps((instance.id, )))

tasks.py

from celery import shared_task


@shared_task(name="create")
def create(data):
    print("Some logic here")

celery.py

import os

from celery import Celery

# Set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'helpdesk.settings')

app = Celery('helpdesk')

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django apps.
app.autodiscover_tasks()

setting.py

CELERY_BROKER_URL = 'redis://redis:6379/0'

Команды, которые я использую для запуска celery и celery-beat:

beat:

celery -A helpdesk beat

celery:

celery -A helpdesk worker -l INFO --pool=solo

Вернуться на верх