How can I schedule a task at time from django frontend?

Question

I'm programming an irrigation system with Django and I use Celery to handle asynchronous tasks.

In my application the user selects the days and hours that he wants to activate the irrigation, this is stored in the database and he can update them at any time.

I know I can schedule tasks like this, but I want to update the time.

from celery.task.schedules import crontab
from celery.decorators import periodic_task

@periodic_task(run_every=crontab(hour=7, minute=30, day_of_week="mon"))
def every_monday_morning():
    print("This runs every Monday morning at 7:30a.m.")

Another way is to use python-crontab

You can create an celery job with spesific date (in seconds). The only thing that you need is calling apply_async with countdown parameter. docs.celeryproject.org/en/latest/userguide/… — Deniz Kaplan
– Deniz Kaplan, Commented Jun 9, 2017 at 11:02

Mark · Accepted Answer · 2019-05-15 20:26:20Z

2

did something similar by having a runtime table listing what I wanted to run and when available to be selected from Django.

Periodic task itself was then scheduled to run every 30 seconds and check if anything requested in this table ready to be executed and called one of the workers.

Edit:as requested by Graham an example

The settings in Django to setup the beat:

CELERY_BEAT_SCHEDULE = {
    'Runtime': {
    'task': 'Runtime',
    'schedule': timedelta(seconds=15),
    },
}

The task calling the model defined in Django (irrigation_requests in this e.g.) that the user would be updating through the front end:

def Runtime():
    #get all from model in state requested
    irrigation_job=Irrigation_Requests.objects.filter(Status__in=['Requested'])

    # Process each requested if flag to do just one thing then you could just test irrigation_job exists      
    for job in irrigation_job:
        ....
        print ("This runs every when requested and depending on job parameters stored in your model can have further scope")

    #update requested this could be done in loop above
    irrigation_job.update(Status='Processed')

edited May 15, 2019 at 20:26

answered Dec 9, 2017 at 16:32

Mark

343 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Graham Over a year ago

Welcome to StackOverflow. Could you provide a code sample here in order to make this answer higher quality?

Mohammed Samir · Accepted Answer · 2022-04-05 20:53:59Z

0

It finally dawned on me after I've seen Mark's answer. So I will do it using a code similar to this:

from celery.task.schedules import crontab
from celery.decorators import periodic_task

@periodic_task(run_every=crontab(minutes="*"))
def run_every_minute():
    now = datetime.datetime.now()
    tasks = TasksModel.objects.all()
    for task in Tasks:
        if task.is_now():
            fn = find_schedual_task(task.name)
            fn()

answered Apr 5, 2022 at 20:53

Mohammed Samir

4254 silver badges10 bronze badges

Collectives™ on Stack Overflow

How can I schedule a task at time from django frontend?

2 Answers 2

1 Comment

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

1 Comment

Comments

Your Answer

Sign up or log in

Post as a guest

Related