dynamically add periodic tasks celery
This works for Celery 4.0.1+ and Python 2.7, and Redis
from celery import Celery
import os, logging
logger = logging.getLogger(__name__)
current_module = __import__(__name__)
CELERY_CONFIG = {
'CELERY_BROKER_URL':
'redis://{}/0'.format(os.environ.get('REDIS_URL', 'localhost:6379')),
'CELERY_TASK_SERIALIZER': 'json',
}
celery = Celery(__name__, broker=CELERY_CONFIG['CELERY_BROKER_URL'])
celery.conf.update(CELERY_CONFIG)
I define a job in the following way:
job = {
'task': 'my_function', # Name of a predefined function
'schedule': {'minute': 0, 'hour': 0} # crontab schedule
'args': [2, 3],
'kwargs': {}
}
I then define a decorator like this:
def add_to_module(f):
setattr(current_module, 'tasks_{}__'.format(f.name), f)
return f
My task is
@add_to_module
def my_function(x, y, **kwargs):
return x + y
Then add a function which adds the task on the fly
def add_task(job):
logger.info("Adding periodic job: %s", job)
if not isinstance(job, dict) and 'task' in jobs:
logger.error("Job {} is ill-formed".format(job))
return False
celery.add_periodic_task(
crontab(**job.get('schedule', {'minute': 0, 'hour': 0})),
get_from_module(job['task']).s(
enterprise_id,
*job.get('args', []),
**job.get('kwargs', {})
),
name = job.get('name'),
expires = job.get('expires')
)
return True
def get_from_module(f):
return getattr(current_module, 'tasks_{}__'.format(f))
After this, you can link the add_task function to a URL, and get them to create tasks out of functions in your current module
for this purpose you can user redBeat. based on redbeat GitHub:
RedBeat is a Celery Beat Scheduler that stores the scheduled tasks and runtime metadata in Redis.
for task creation:
import tasks #celery defined task class
from redbeat import RedBeatSchedulerEntry as Entry
entry = Entry(f'urlCheck_{key}', 'tasks.urlSpeed', repeat, args=['GET', url, timeout, key], app=tasks.app)
entry.save()
entry.key
delete task:
import tasks #celery defined task class
from redbeat import RedBeatSchedulerEntry as Entry
entry = Entry.from_key(key, app=tasks.app) #key from previous step
entry.delete()
and there is an example that you can use: https://github.com/hamedsh/redBeat_example