Celery+Docker+Django -- Getting tasks to work

Question

I've been trying to learn Celery over the past week and adding it to my project that uses Django and Docker-Compose. I am having a hard time understanding how to get it to work; my issue is that I can't seem to get uploading to my database to work when using tasks. The upload function, insertIntoDatabase, was working fine before without any involvement with Celery but now uploading doesn't work. Indeed, when I try to upload, my website tells me too quickly that the upload was successful, but then nothing actually gets uploaded.

The server is started up with docker-compose up, which will make migrations, perform a migrate, collect static files, update requirements, and then start the server. This is all done using pavement.py; the command in the Dockerfile is CMD paver docker_run. At no point is a Celery worker explicitly started; should I be doing that? If so, how?

This is the way I'm calling the upload function in views.py:

insertIntoDatabase.delay(datapoints, user, description)

The upload function is defined in a file named databaseinserter.py. The following decorator was used for insertIntoDatabase:

@shared_task(bind=True, name="database_insert", base=DBTask)

Here is the definition of the DBTask class in celery.py:

class DBTask(Task):
     abstract = True

     def on_failure(self, exc, *args, **kwargs):
        raise exc

I am not really sure what to write for tasks.py. Here is what I was left with by a former co-worker just before I picked up from where he left off:

from celery.decorators import task
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@task(name="database_insert")
def database_insert(data):

And here are the settings I used to configure Celery (settings.py):

BROKER_TRANSPORT = 'redis'
_REDIS_LOCATION = 'redis://{}:{}'.format(os.environ.get("REDIS_PORT_6379_TCP_ADDR"), os.environ.get("REDIS_PORT_6379_TCP_PORT"))
BROKER_URL = _REDIS_LOCATION + '/0'
CELERY_RESULT_BACKEND = _REDIS_LOCATION + '/1'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ENABLE_UTC = True
CELERY_TIMEZONE = "UTC"

Now, I'm guessing that database_insert in tasks.py shouldn't be empty, but what should go there instead? Also, it doesn't seem like anything in tasks.py happens anyway--when I added some logging statements to see if tasks.py was at least being run, nothing actually ended up getting logged, making me think that tasks.py isn't even being run. How do I properly make my upload function into a task?

Answer 1

You're not too far off from getting this working, I think.

First, I'd recommend that you do try to keep your Celery tasks and your business logic separate. So, for example, it probably makes good sense to have the business logic involved with inserting your data into your DB in the insertIntoDatabase function, and then separately create a Celery task, perhaps name insert_into_db_task, that takes in your args as plain python objects (important) and calls the aforementioned insertIntoDatabase function with those args to actually complete the DB insertion.

Code for that example might looks like this:

my_app/tasks/insert_into_db.py

from celery.decorators import task
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@task()
def insert_into_db_task(datapoints, user, description):
    from my_app.services import insertIntoDatabase
    insertIntoDatabase(datapoints, user, description)

my_app/services/insertIntoDatabase.py

def insertIntoDatabase(datapoints, user, description):
    """Note that this function is not a task, by design"""

    # do db insertion stuff

my_app/views/insert_view.py

from my_app.tasks import insert_into_db_task

def simple_insert_view_func(request, args, kwargs):
    # start handling request, define datapoints, user, description

    # next line creates the **task** which will later do the db insertion
    insert_into_db_task.delay(datapoints, user, description)
    return Response(201)

The app structure I'm implying is just how I would do it and isn't required. Note also that you can probably use @task() straight up and not define any args for it. Might simplify things for you.

Does that help? I like to keep my tasks light and fluffy. They mostly just do jerk proofing (make sure the involved objs exist in DB, for instance), tweak what happens if the task fails (retry later? abort task? etc.), logging, and otherwise they execute business logic that lives elsewhere.

Also, in case it's not obvious, you do need to be running celery somewhere so that there are workers to actually process the tasks that your view code are creating. If you don't run celery somewhere then your tasks will just stack up in the queue and never get processed (and so your DB insertions will never happen).