Celery Commands Etc

This summary is not available. Please click here to view the post.

Celery Tasks Parallel and Chained Execution Workflow

In celery, it is very easy to chain and parallelize execution of tasks, e.g. to satisfy this example workflow...

      _____
     |task1|      1. exec task1
    /      \
 __/__    __\__
|task2|  |task3|  2. parallel task1 & task2
   \        /
    \ _____/
     |task4|      3. exec task4 when both task1 & task2 are done

from celery import chain, group, chord
from orchestrator.tasks.host_tasks import acquire_hosts, prepare_hosts, return_hosts
from orchestrator.tasks.job_tasks import start_job, stop_job
from orchestrator.tasks.vcenter_tasks import acquire_vcenter, return_vcenter


def start_performance_run_workflow(job_id):
    print('Starting performance run workflows for job %s' % job_id)
    # The effect of this chain, is that only the first one is routed
    # to the "regular.priority" queue
    # subsequent tasks are routed to "celery" default queue
    workflow = chain(
        acquire_hosts.si(job_id),
        chord(
            (prepare_hosts.si(job_id), acquire_vcenter.si(job_id)),
            start_job.si(job_id)
        )
    ).apply_async(queue="regular.priority")

Async Background Tasks with Celery on Django

From this post and this post and this post

1. Install Redis - got from here
$ brew install redis
$ ln -sfv /usr/local/opt/redis/*.plist ~/Library/LaunchAgents
$ launchctl load ~/Library/LaunchAgents/homebrew.mxcl.redis.plist
$ redis-cli PING
PONG

2. Install celery with redis support bundle
$ pip3 install -U "celery[redis]"

3. Create celery.py in django project
$ vi sample_project/sample_project/celery.py
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'sample_project.settings')

app = Celery('sample_project')

# Using a string here means the worker don't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()


@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

4. Import celery app in __init__.py
from __future__ import absolute_import, unicode_literals

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = ['celery_app']

5. Add config in settings.py
$ vi sample_project/sample_project/settings.py
# celery
CELERY_BROKER_URL = 'redis://localhost:6379/0'

6. Create app
$ python3 manage.py startapp testapp
$ vi sample_project/sample_project/settings.py
INSTALLED_APPS = (
    # (...)
    'testapp',
)

7. Create task
$ vi sample_project/sample_project/testapp/tasks.py
from __future__ import absolute_import

from celery import shared_task

@shared_task
def test(param):
    return 'The test task executed with argument "%s" ' % param

8. Testing

Run server
$ python3 manage.py runserver

Run worker
$ celery -A sample_project worker --loglevel=info

From another terminal...
$ python3 manage.py shell
Python 3.6.2 (default, Jul 17 2017, 16:44:45)
[GCC 4.2.1 Compatible Apple LLVM 8.1.0 (clang-802.0.42)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
(InteractiveConsole)
>>> from testapp.tasks import test
>>> test.delay('This is a test')


From the worker terminal
2017-07-26 00:50:54,529: INFO/MainProcess] Received task: testapp.tasks.test[4f1ab1d4-645d-480b-a1b5-6e2887e34b52]
[2017-07-26 00:50:54,533: INFO/ForkPoolWorker-2] Task testapp.tasks.test[4f1ab1d4-645d-480b-a1b5-6e2887e34b52] succeeded in 0.0010309600038453937s: 'The test task executed with argument "This is a test" '

9. For results backend using Django ORM

a. $ pip3 install django-celery-results

b. Install app in settings.py
INSTALLED_APPS = (
    ...,
    'django_celery_results',
)

c. Create celery database tables
$ python3 manage.py migrate django_celery_results

d. Configure Celery to use the django-celery-results backend.
In settings.py...
CELERY_RESULT_BACKEND = 'django-db'