How to dynamically add / remove periodic tasks

[Jamie Forrest]

Received: by 10.224.197.7 with SMTP id ei7mr107359qab.16.1334823252763;
Thu, 19 Apr 2012 01:14:12 -0700 (PDT)
X-BeenThere: celery-users-/***@public.gmane.org
Received: by 10.224.33.144 with SMTP id h16ls2199684qad.4.gmail; Thu, 19 Apr
2012 01:14:11 -0700 (PDT)
Received: by 10.224.97.134 with SMTP id l6mr660407qan.6.1334823251405;
Thu, 19 Apr 2012 01:14:11 -0700 (PDT)
Received: by 10.224.8.211 with SMTP id i19msqai;
Wed, 18 Apr 2012 16:30:44 -0700 (PDT)
Received: by 10.52.76.198 with SMTP id m6mr446006vdw.12.1334791844552; Wed, 18
Apr 2012 16:30:44 -0700 (PDT)
Received: by z38g2000vbu.googlegroups.com with HTTP; Wed, 18 Apr 2012 16:30:44
-0700 (PDT)
User-Agent: G2/1.0
X-HTTP-UserAgent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_3)
AppleWebKit/535.19 (KHTML, like Gecko) Chrome/18.0.1025.162 Safari/535.19,gzip(gfe)
X-Original-Sender: jamie-***@public.gmane.org
X-Original-Authentication-Results: ls.google.com; spf=pass (google.com: domain
of jamie-***@public.gmane.org designates internal as permitted sender)
smtp.mail=jamie-***@public.gmane.org; dkim=pass header.i=@jamieforrest.com
Precedence: list
Mailing-list: list celery-users-/***@public.gmane.org; contact celery-users+owners-/***@public.gmane.org
List-ID: <celery-users.googlegroups.com>
X-Google-Group-Id: 634191933412
List-Post: <http://groups.google.com/group/celery-users/post?hl=en_US>, <mailto:celery-users-/***@public.gmane.org>
List-Help: <http://groups.google.com/support/?hl=en_US>, <mailto:celery-users+help-/***@public.gmane.org>
List-Archive: <http://groups.google.com/group/celery-users?hl=en_US>
Sender: celery-users-/***@public.gmane.org
List-Subscribe: <http://groups.google.com/group/celery-users/subscribe?hl=en_US>,
<mailto:celery-users+subscribe-/***@public.gmane.org>
List-Unsubscribe: <http://groups.google.com/group/celery-users/subscribe?hl=en_US>,
<mailto:googlegroups-manage+634191933412+unsubscribe-/***@public.gmane.org>
X-Gm-Message-State: ALoCoQnP53Qb2x6C4BXSMfR5m6IUvP0euoGJ7E6pqeZ3MsgYbbs3Uyfw9YtbebiYX0KXDs3V9TqY
Archived-At: <http://permalink.gmane.org/gmane.comp.python.amqp.celery.user/2300>

Apologies for the cross-posting to Stack Overflow (http://
stackoverflow.com/questions/10194975/how-to-dynamically-add-remove-
periodic-tasks-to-celery-celerybeat). I haven't had any luck getting a
response over there so I figured I'd try here.

If I have a function defined as follows:

def add(x,y):
return x+y

Is there a way to dynamically add this function as a celery
PeriodicTask and kick it off at runtime? I'd like to be able to do
something like (pseudocode):

some_unique_task_id = celery.beat.schedule_task(add,
run_every=crontab(minute="*/30"))
celery.beat.start(some_unique_task_id)

I would also want to stop or remove that task dynamically with
something like (pseudocode):

celery.beat.remove_task(some_unique_task_id)
-or-
celery.beat.stop(some_unique_task_id)

FYI I am not using djcelery, which lets you manage periodic tasks via
the django admin.

Thanks,
Jamie Forrest

[Ask Solem]

Post by Jamie Forrest
Apologies for the cross-posting to Stack Overflow (http://
stackoverflow.com/questions/10194975/how-to-dynamically-add-remove-
periodic-tasks-to-celery-celerybeat). I haven't had any luck getting a
response over there so I figured I'd try here.
return x+y
Is there a way to dynamically add this function as a celery
PeriodicTask and kick it off at runtime? I'd like to be able to do
some_unique_task_id = celery.beat.schedule_task(add,
run_every=crontab(minute="*/30"))
celery.beat.start(some_unique_task_id)
I would also want to stop or remove that task dynamically with
celery.beat.remove_task(some_unique_task_id)
-or-
celery.beat.stop(some_unique_task_id)
FYI I am not using djcelery, which lets you manage periodic tasks via
the django admin.

No, I'm sorry, this is not possible with the regular celerybeat.

But it's easily extensible to do what you want, e.g. the django-celery
scheduler is just a subclass reading and writing the schedule to the database
(with some optimizations on top).

Also you can use the django-celery scheduler even for non-Django projects.

Something like this:

- Install django + django-celery:

$ pip install -U django django-celery

- Add the following settings to your celeryconfig:

DATABASES = {"default": {"NAME": "celerybeat.db",
"ENGINE": "django.db.backends.sqlite3"}}
INSTALLED_APPS = ("djcelery", )

- Create the database tables:

$ PYTHONPATH=. django-admin.py syncdb --settings=celeryconfig

- Start celerybeat with the database scheduler:
$ PYTHONPATH=. django-admin.py celerybeat --settings=celeryconfig \
-S djcelery.schedulers.DatabaseScheduler


Also there's the `djcelerymon` command which can be used for non-Django projects
to start celerycam and a Django Admin webserver in the same process, you can
use that to also edit your periodic tasks in a nice web interface:

$ djcelerymon


(Note for some reason djcelerymon can't be stopped using Ctrl+C, you
have to use Ctrl+Z + kill %1)

[Jean-Mark]

Is there a way that this can be done using djcelery but not using the Admin
interface? I'm able to add tasks successfully but even when I delete them
from the database beat still supplies them to the workers. How can I stop
that?

Post by Ask Solem

Post by Jamie Forrest
Apologies for the cross-posting to Stack Overflow (http://
stackoverflow.com/questions/10194975/how-to-dynamically-add-remove-
periodic-tasks-to-celery-celerybeat). I haven't had any luck getting a
response over there so I figured I'd try here.
return x+y
Is there a way to dynamically add this function as a celery
PeriodicTask and kick it off at runtime? I'd like to be able to do
some_unique_task_id = celery.beat.schedule_task(add,
run_every=crontab(minute="*/30"))
celery.beat.start(some_unique_task_id)
I would also want to stop or remove that task dynamically with
celery.beat.remove_task(some_unique_task_id)
-or-
celery.beat.stop(some_unique_task_id)
FYI I am not using djcelery, which lets you manage periodic tasks via
the django admin.

No, I'm sorry, this is not possible with the regular celerybeat.
But it's easily extensible to do what you want, e.g. the django-celery
scheduler is just a subclass reading and writing the schedule to the database
(with some optimizations on top).
Also you can use the django-celery scheduler even for non-Django projects.
$ pip install -U django django-celery
DATABASES = {"default": {"NAME": "celerybeat.db",
"ENGINE": "django.db.backends.sqlite3"}}
INSTALLED_APPS = ("djcelery", )
$ PYTHONPATH=. django-admin.py syncdb --settings=celeryconfig
$ PYTHONPATH=. django-admin.py celerybeat --settings=celeryconfig \
-S djcelery.schedulers.DatabaseScheduler
Also there's the `djcelerymon` command which can be used for non-Django projects
to start celerycam and a Django Admin webserver in the same process, you can
$ djcelerymon
(Note for some reason djcelerymon can't be stopped using Ctrl+C, you
have to use Ctrl+Z + kill %1)
--
Ask Solem
twitter.com/asksol | [phone]

[Jean-Mark]

Just replying since I found a solution for this. When you're deleting a
Periodic Task, first set it's enabled property to false FIRST. Then save it
then delete it. Otherwise the broker will hold on to it it seems.

Post by Jean-Mark
Is there a way that this can be done using djcelery but not using the
Admin interface? I'm able to add tasks successfully but even when I delete
them from the database beat still supplies them to the workers. How can I
stop that?

Post by Ask Solem

Post by Jamie Forrest
Apologies for the cross-posting to Stack Overflow (http://
stackoverflow.com/questions/10194975/how-to-dynamically-add-remove-
periodic-tasks-to-celery-celerybeat). I haven't had any luck getting a
response over there so I figured I'd try here.
return x+y
Is there a way to dynamically add this function as a celery
PeriodicTask and kick it off at runtime? I'd like to be able to do
some_unique_task_id = celery.beat.schedule_task(add,
run_every=crontab(minute="*/30"))
celery.beat.start(some_unique_task_id)
I would also want to stop or remove that task dynamically with
celery.beat.remove_task(some_unique_task_id)
-or-
celery.beat.stop(some_unique_task_id)
FYI I am not using djcelery, which lets you manage periodic tasks via
the django admin.

No, I'm sorry, this is not possible with the regular celerybeat.
But it's easily extensible to do what you want, e.g. the django-celery
scheduler is just a subclass reading and writing the schedule to the database
(with some optimizations on top).
Also you can use the django-celery scheduler even for non-Django projects.
$ pip install -U django django-celery
DATABASES = {"default": {"NAME": "celerybeat.db",
"ENGINE": "django.db.backends.sqlite3"}}
INSTALLED_APPS = ("djcelery", )
$ PYTHONPATH=. django-admin.py syncdb --settings=celeryconfig
$ PYTHONPATH=. django-admin.py celerybeat --settings=celeryconfig \
-S djcelery.schedulers.DatabaseScheduler
Also there's the `djcelerymon` command which can be used for non-Django projects
to start celerycam and a Django Admin webserver in the same process, you can
$ djcelerymon
(Note for some reason djcelerymon can't be stopped using Ctrl+C, you
have to use Ctrl+Z + kill %1)
--
Ask Solem
twitter.com/asksol | [phone]

[Jean-Mark]

Here's a proper solution for this. Confirmed working, In my scenario, I
sub-classed Periodic Task and created a model out of it since I can add
other fields to the model as I need and also so I could add the "terminate"
method. You have to set the periodic task's enabled property to False and
save it before you delete it. The whole subclassing is not a must, the
schedule_every method is the one that really does the work. When you're
ready to terminate you task (if you didn't subclass it) you can just use
PeriodicTask.objects.filter(name=...) to search for your task, disable it,
then delete it.

Hope this helps!

from djcelery.models import PeriodicTask, IntervalSchedule
from datetime import datetime

class TaskScheduler(models.Model):

periodic_task = models.ForeignKey(PeriodicTask)

@staticmethod
def schedule_every(task_name, period, every, args=None, kwargs=None):
""" schedules a task by name every "every" "period". So an example call
would be:
TaskScheduler('mycustomtask', 'seconds', 30, [1,2,3])
that would schedule your custom task to run every 30 seconds with
the arguments 1,2 and 3 passed to the actual task.
"""
permissible_periods = ['days', 'hours', 'minutes', 'seconds']
if period not in permissible_periods:
raise Exception('Invalid period specified')
# create the periodic task and the interval
ptask_name = "%s_%s" % (task_name, datetime.datetime.now()) #
create some name for the period task
interval_schedules = IntervalSchedule.objects.filter(period=period,
every=every)
if interval_schedules: # just check if interval schedules exist
like that already and reuse em
interval_schedule = interval_schedules[0]
else: # create a brand new interval schedule
interval_schedule = IntervalSchedule()
interval_schedule.every = every # should check to make sure
this is a positive int
interval_schedule.period = period
interval_schedule.save()
ptask = PeriodicTask(name=ptask_name, task=task_name,
interval=interval_schedule)
if args:
ptask.args = args
if kwargs:
ptask.kwargs = kwargs
ptask.save()
return TaskScheduler.objects.create(periodic_task=ptask)

def stop(self):
""" pauses the task
ptask = self.periodic_task
ptask.enabled = False
ptask.save()

def start(self):
ptask = self.periodic_task
ptask.enabled = True
ptask.save()

def terminate(self):
self.stop()
ptask = self.periodic_task
self.delete()
ptask.delete()

Post by Jean-Mark
Just replying since I found a solution for this. When you're deleting a
Periodic Task, first set it's enabled property to false FIRST. Then save it
then delete it. Otherwise the broker will hold on to it it seems.

Post by Jean-Mark
Is there a way that this can be done using djcelery but not using the
Admin interface? I'm able to add tasks successfully but even when I delete
them from the database beat still supplies them to the workers. How can I
stop that?

Post by Ask Solem

Post by Jamie Forrest
Apologies for the cross-posting to Stack Overflow (http://
stackoverflow.com/questions/10194975/how-to-dynamically-add-remove-
periodic-tasks-to-celery-celerybeat). I haven't had any luck getting a
response over there so I figured I'd try here.
return x+y
Is there a way to dynamically add this function as a celery
PeriodicTask and kick it off at runtime? I'd like to be able to do
some_unique_task_id = celery.beat.schedule_task(add,
run_every=crontab(minute="*/30"))
celery.beat.start(some_unique_task_id)
I would also want to stop or remove that task dynamically with
celery.beat.remove_task(some_unique_task_id)
-or-
celery.beat.stop(some_unique_task_id)
FYI I am not using djcelery, which lets you manage periodic tasks via
the django admin.

No, I'm sorry, this is not possible with the regular celerybeat.
But it's easily extensible to do what you want, e.g. the django-celery
scheduler is just a subclass reading and writing the schedule to the database
(with some optimizations on top).
Also you can use the django-celery scheduler even for non-Django projects.
$ pip install -U django django-celery
DATABASES = {"default": {"NAME": "celerybeat.db",
"ENGINE": "django.db.backends.sqlite3"}}
INSTALLED_APPS = ("djcelery", )
$ PYTHONPATH=. django-admin.py syncdb --settings=celeryconfig
$ PYTHONPATH=. django-admin.py celerybeat --settings=celeryconfig \
-S djcelery.schedulers.DatabaseScheduler
Also there's the `djcelerymon` command which can be used for non-Django projects
to start celerycam and a Django Admin webserver in the same process, you can
$ djcelerymon
(Note for some reason djcelerymon can't be stopped using Ctrl+C, you
have to use Ctrl+Z + kill %1)
--
Ask Solem
twitter.com/asksol | [phone]

[Nilesh Sutar]

Hi Jean,
Thank you for your approch.
The TaskScheduler Model doesn't support the weekly task and the task
interval doesn't supports the other periods other than listed.
Do you have clue to support execution of task weekly, monthly using Django
Celery?

Post by Jean-Mark
Here's a proper solution for this. Confirmed working, In my scenario, I
sub-classed Periodic Task and created a model out of it since I can add
other fields to the model as I need and also so I could add the "terminate"
method. You have to set the periodic task's enabled property to False and
save it before you delete it. The whole subclassing is not a must, the
schedule_every method is the one that really does the work. When you're
ready to terminate you task (if you didn't subclass it) you can just use
PeriodicTask.objects.filter(name=...) to search for your task, disable it,
then delete it.
Hope this helps!
from djcelery.models import PeriodicTask, IntervalSchedule
from datetime import datetime
periodic_task = models.ForeignKey(PeriodicTask)
@staticmethod
""" schedules a task by name every "every" "period". So an example
TaskScheduler('mycustomtask', 'seconds', 30, [1,2,3])
that would schedule your custom task to run every 30 seconds with
the arguments 1,2 and 3 passed to the actual task.
"""
permissible_periods = ['days', 'hours', 'minutes', 'seconds']
raise Exception('Invalid period specified')
# create the periodic task and the interval
ptask_name = "%s_%s" % (task_name, datetime.datetime.now()) #
create some name for the period task
interval_schedules =
IntervalSchedule.objects.filter(period=period, every=every)
if interval_schedules: # just check if interval schedules exist
like that already and reuse em
interval_schedule = interval_schedules[0]
else: # create a brand new interval schedule
interval_schedule = IntervalSchedule()
interval_schedule.every = every # should check to make sure
this is a positive int
interval_schedule.period = period
interval_schedule.save()
ptask = PeriodicTask(name=ptask_name, task=task_name,
interval=interval_schedule)
ptask.args = args
ptask.kwargs = kwargs
ptask.save()
return TaskScheduler.objects.create(periodic_task=ptask)
""" pauses the task
ptask = self.periodic_task
ptask.enabled = False
ptask.save()
ptask = self.periodic_task
ptask.enabled = True
ptask.save()
self.stop()
ptask = self.periodic_task
self.delete()
ptask.delete()

Post by Jean-Mark
Just replying since I found a solution for this. When you're deleting a
Periodic Task, first set it's enabled property to false FIRST. Then save it
then delete it. Otherwise the broker will hold on to it it seems.

Post by Jean-Mark
Is there a way that this can be done using djcelery but not using the
Admin interface? I'm able to add tasks successfully but even when I delete
them from the database beat still supplies them to the workers. How can I
stop that?

Post by Ask Solem

Post by Jamie Forrest
Apologies for the cross-posting to Stack Overflow (http://
stackoverflow.com/questions/10194975/how-to-dynamically-add-remove-
periodic-tasks-to-celery-celerybeat). I haven't had any luck getting a
response over there so I figured I'd try here.
return x+y
Is there a way to dynamically add this function as a celery
PeriodicTask and kick it off at runtime? I'd like to be able to do
some_unique_task_id = celery.beat.schedule_task(add,
run_every=crontab(minute="*/30"))
celery.beat.start(some_unique_task_id)
I would also want to stop or remove that task dynamically with
celery.beat.remove_task(some_unique_task_id)
-or-
celery.beat.stop(some_unique_task_id)
FYI I am not using djcelery, which lets you manage periodic tasks via
the django admin.

No, I'm sorry, this is not possible with the regular celerybeat.
But it's easily extensible to do what you want, e.g. the django-celery
scheduler is just a subclass reading and writing the schedule to the database
(with some optimizations on top).
Also you can use the django-celery scheduler even for non-Django projects.
$ pip install -U django django-celery
DATABASES = {"default": {"NAME": "celerybeat.db",
"ENGINE": "django.db.backends.sqlite3"}}
INSTALLED_APPS = ("djcelery", )
$ PYTHONPATH=. django-admin.py syncdb --settings=celeryconfig
$ PYTHONPATH=. django-admin.py celerybeat --settings=celeryconfig \
-S djcelery.schedulers.DatabaseScheduler
Also there's the `djcelerymon` command which can be used for non-Django projects
to start celerycam and a Django Admin webserver in the same process, you can
$ djcelerymon
(Note for some reason djcelerymon can't be stopped using Ctrl+C, you
have to use Ctrl+Z + kill %1)
--
Ask Solem
twitter.com/asksol | [phone]

[RON MICHAEL]

Thanks for this! Do you by any chance know how to get the current schedule
that's being run by cron? Also, what happens if crons with the same time
are created?

Post by Jean-Mark
Here's a proper solution for this. Confirmed working, In my scenario, I
sub-classed Periodic Task and created a model out of it since I can add
other fields to the model as I need and also so I could add the "terminate"
method. You have to set the periodic task's enabled property to False and
save it before you delete it. The whole subclassing is not a must, the
schedule_every method is the one that really does the work. When you're
ready to terminate you task (if you didn't subclass it) you can just use
PeriodicTask.objects.filter(name=...) to search for your task, disable it,
then delete it.
Hope this helps!
from djcelery.models import PeriodicTask, IntervalSchedule
from datetime import datetime
periodic_task = models.ForeignKey(PeriodicTask)
@staticmethod
""" schedules a task by name every "every" "period". So an example
TaskScheduler('mycustomtask', 'seconds', 30, [1,2,3])
that would schedule your custom task to run every 30 seconds with
the arguments 1,2 and 3 passed to the actual task.
"""
permissible_periods = ['days', 'hours', 'minutes', 'seconds']
raise Exception('Invalid period specified')
# create the periodic task and the interval
ptask_name = "%s_%s" % (task_name, datetime.datetime.now()) #
create some name for the period task
interval_schedules =
IntervalSchedule.objects.filter(period=period, every=every)
if interval_schedules: # just check if interval schedules exist
like that already and reuse em
interval_schedule = interval_schedules[0]
else: # create a brand new interval schedule
interval_schedule = IntervalSchedule()
interval_schedule.every = every # should check to make sure
this is a positive int
interval_schedule.period = period
interval_schedule.save()
ptask = PeriodicTask(name=ptask_name, task=task_name,
interval=interval_schedule)
ptask.args = args
ptask.kwargs = kwargs
ptask.save()
return TaskScheduler.objects.create(periodic_task=ptask)
""" pauses the task
ptask = self.periodic_task
ptask.enabled = False
ptask.save()
ptask = self.periodic_task
ptask.enabled = True
ptask.save()
self.stop()
ptask = self.periodic_task
self.delete()
ptask.delete()

Post by Jean-Mark
Just replying since I found a solution for this. When you're deleting a
Periodic Task, first set it's enabled property to false FIRST. Then save it
then delete it. Otherwise the broker will hold on to it it seems.

Post by Jean-Mark
Is there a way that this can be done using djcelery but not using the
Admin interface? I'm able to add tasks successfully but even when I delete
them from the database beat still supplies them to the workers. How can I
stop that?

Post by Ask Solem

Post by Jamie Forrest
Apologies for the cross-posting to Stack Overflow (http://
stackoverflow.com/questions/10194975/how-to-dynamically-add-remove-
periodic-tasks-to-celery-celerybeat). I haven't had any luck getting a
response over there so I figured I'd try here.
return x+y
Is there a way to dynamically add this function as a celery
PeriodicTask and kick it off at runtime? I'd like to be able to do
some_unique_task_id = celery.beat.schedule_task(add,
run_every=crontab(minute="*/30"))
celery.beat.start(some_unique_task_id)
I would also want to stop or remove that task dynamically with
celery.beat.remove_task(some_unique_task_id)
-or-
celery.beat.stop(some_unique_task_id)
FYI I am not using djcelery, which lets you manage periodic tasks via
the django admin.

No, I'm sorry, this is not possible with the regular celerybeat.
But it's easily extensible to do what you want, e.g. the django-celery
scheduler is just a subclass reading and writing the schedule to the database
(with some optimizations on top).
Also you can use the django-celery scheduler even for non-Django projects.
$ pip install -U django django-celery
DATABASES = {"default": {"NAME": "celerybeat.db",
"ENGINE": "django.db.backends.sqlite3"}}
INSTALLED_APPS = ("djcelery", )
$ PYTHONPATH=. django-admin.py syncdb --settings=celeryconfig
$ PYTHONPATH=. django-admin.py celerybeat --settings=celeryconfig \
-S djcelery.schedulers.DatabaseScheduler
Also there's the `djcelerymon` command which can be used for non-Django projects
to start celerycam and a Django Admin webserver in the same process, you can
$ djcelerymon
(Note for some reason djcelerymon can't be stopped using Ctrl+C, you
have to use Ctrl+Z + kill %1)
--
Ask Solem
twitter.com/asksol | [phone]