41👍
✅
What’s wrong with the example from the docs?
from celery.task import PeriodicTask
from clickmuncher.messaging import process_clicks
from datetime import timedelta
class ProcessClicksTask(PeriodicTask):
run_every = timedelta(minutes=30)
def run(self, **kwargs):
process_clicks()
You could write the same task using a decorator:
from celery.task.schedules import crontab
from celery.task import periodic_task
@periodic_task(run_every=crontab(minute="*/30"))
def process_clicks():
....
The decorator syntax simply allows you to turn an existing function/task into a periodic task without modifying them directly.
For the tasks to be executed celerybeat must be running.
👤arie
Source:stackexchange.com