Django Celery: Clocked task is not running

Question

In a Django app, I have a form that schedules an email to be sent out. It has four fields: name, email, body, send_date. I want to dynamically create a Celery task (email) to run another Celery task at the designated time.

I have been able to send out the email at regular intervals (every 30 seconds) based on the form using the following code:

schedule, _ = IntervalSchedule.objects.update_or_create(every=30, period=IntervalSchedule.SECONDS)
@shared_task(name="schedule_interval_email")
def schedule_email_interval(name, email, body, send_date):
    PeriodicTask.objects.update_or_create(
        defaults={
            "interval": schedule,
            "task": "email_task"
        },
        name="Send message at interval",
        args=json.dumps(['name', 'test@test.com', 'body']),
    )

However, when I have tried to schedule a task to run at a specific time (3 minutes later from the current time) via ClockedSchedule, Celery beat records the tasks and saves all the relevant settings. The task appears active in the Django admin area. However, the email never actually gets sent.

clocked = ClockedSchedule.objects.create(clocked_time=datetime.now() + timedelta(minutes=3))
@shared_task(name="schedule_clock_email")
def schedule_email_clocked(name, email, body, send_date):
    PeriodicTask.objects.create(
        clocked=clocked,
        name="Send message at specific date/time",
        task="email_task",
        one_off=True,
        args=json.dumps(['name', 'test@test.com', 'body']),
    )

I eventually want to dynamically set the clocked field based on the datetime the user enters into the form, so the current code is just trying to test out the way Celery works. I think I'm missing something about how this works, though. Any thoughts would be greatly appreciated.

Answer 1

I guess it's the time zone.

If your configuration does not take effect.The celery database stores UTC time by default.

You can verify this by executing the following code.

In Django's shell environment

import datetime
import json
from ProjectName.celerytest import test

execution_time = datetime.datetime.utcnow() + datetime.timedelta(minutes=2)

schedule, created = ClockedSchedule.objects.get_or_create(clocked_time=execution_time)


PeriodicTask.objects.create(clocked=schedule,one_off=True,name="test input",task="projectName.celerytest.test",args=json.dumps(['112233']))