Last active
July 13, 2016 03:12
-
-
Save mydreambei-ai/da854c170b3f195ea90afafd6bd35d98 to your computer and use it in GitHub Desktop.
celery create periodic tasks in different ways
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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