Skip to content

Instantly share code, notes, and snippets.

@mydreambei-ai
Last active July 13, 2016 03:12
Show Gist options
  • Save mydreambei-ai/da854c170b3f195ea90afafd6bd35d98 to your computer and use it in GitHub Desktop.
Save mydreambei-ai/da854c170b3f195ea90afafd6bd35d98 to your computer and use it in GitHub Desktop.
celery create periodic tasks in different ways
from celery.task import periodic_task
from datetime import timedelta
# first way
@periodic_task(run_every=timedelta(seconds=10), exchange="default", routing_key="default")
def every_10_seconds():
print("This is test")
return 1
# second way
from celery.task import PeriodicTask
class MyPeriodTask(PeriodicTask):
run_every = timedelta(seconds=6)
options = {"exchange": "default", "routing_key": "default"}
name = "xxxxx"
def run(self):
print("this is test")
return 2
# 3th way
def run_period_task():
"""
this way must define CELERYBEAT_SCHEDULE in config file like:
CELERYBEAT_SCHEDULE = {
'run_period_task': {
'task': '<project>.tasks.run_period_task',
'schedule': timedelta(seconds=30)
},
}
# http://celery.readthedocs.io/en/latest/userguide/periodic-tasks.html
"""
print("this is test")
return 3
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment