


Tasks are the building blocks of Celery applications.


Every task class has a unique name, and this name is referenced in messages so that the worker can find the right function to execute.

A task message does not disappear until the message has been acknowledged by a worker. A worker can reserve many messages in advance and even if the worker is killed – caused by power failure or otherwise – the message will be redelivered to another worker.

Ideally task functions should be idempotent, which means that the function will not cause unintented effects even if called multiple times with the same arguments. Since the worker cannot detect if your tasks are idempotent, the default behavior is to acknowledge the message in advance, before it’s executed, so that a task that has already been started is never executed again..


In this chapter you will learn all about defining tasks, and this is the table of contents:



from .models import User

def create_user(username, password):
    User.objects.create(username=username, password=password)


def create_user(username, password):
    User.objects.create(username=username, password=password)


Every task must have a unique name, and a new name will be generated out of the function name if a custom name is not provided.

For example:

>>> def add(x, y):
...     return x + y


A best practice is to use the module name as a namespace, this way names won’t collide if there’s already a task with that name defined in another module.

>>> @app.task(name='tasks.add')
>>> def add(x, y):
...     return x + y

You can tell the name of the task by investigating its name attribute:


Which is exactly the name that would have been generated anyway, if the module name is “”:

def add(x, y):
    return x + y

>>> from tasks import add

Automatic naming and relative imports

Relative imports and automatic name generation do not go well together, so if you’re using relative imports you should set the name explicitly.

For example if the client imports the module “myapp.tasks” as ”.tasks”, and the worker imports the module as “myapp.tasks”, the generated names won’t match and an NotRegistered error will be raised by the worker.


INSTALLED_APPS = ['project.myapp']


>>> from project.myapp.tasks import mytask   # << GOOD



>>> from project.myapp.tasks import mytask

>>> from myapp.tasks import mytask

So for this reason you must be consistent in how you import modules, which is also a Python best practice.

Similarly, you should not use old-style relative imports:

from module import foo   # BAD!

from proj.module import foo  # GOOD!

New-style relative imports are fine and can be used:

from .module import foo  # GOOD!


def add(x, y):
    return x + y


request contains information and state related to the executing task.

The request defines the following attributes:

id:The unique id of the executing task.
group:The unique id a group, if this task is a member.
args:Positional arguments.
kwargs:Keyword arguments.
retries:How many times the current task has been retried. An integer starting at 0.
is_eager:Set to True if the task is executed locally in the client, and not by a worker.
eta:The original ETA of the task (if any). This is in UTC time (depending on the CELERY_ENABLE_UTC setting).
logfile:The file the worker logs to. See Logging.
loglevel:The current log level used.
hostname:Hostname of the worker instance executing the task.
delivery_info:Additional message delivery information. This is a mapping containing the exchange and routing key used to deliver this task. Used by e.g. retry() to resend the task to the same destination queue. Availability of keys in this dict depends on the message broker used.
 This flag is set to true if the task was not executed by the worker.
callbacks:A list of subtasks to be called if this task returns successfully.
errback:A list of subtasks to be called if this task fails.
utc:Set to true the caller has utc enabled (CELERY_ENABLE_UTC).

New in version 3.1.

headers:Mapping of message headers (may be None).
reply_to:Where to send reply to (queue name).

An example task accessing information in the context is:

def dump_context(self, x, y):






from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

    return x + y

Celery uses the standard Python logger library, for which documentation can be found in the logging module.



The worker will not update the redirection if you create a logger instance somewhere in your task or task module.


import sys

logger = get_task_logger(__name__)

def add(self, x, y):
    old_outs = sys.stdout, sys.stderr
    rlevel =
    try:, rlevel)
        return x + y
        sys.stdout, sys.stderr = old_outs



When you call retry it will send a new message, using the same task-id, and it will take care to make sure the message is delivered to the same queue as the originating task.


Here’s an example using retry:

def send_twitter_status(self, oauth, tweet):
        twitter = Twitter(oauth)
    except (Twitter.FailWhaleError, Twitter.LoginError) as exc:
        raise self.retry(exc=exc)


The retry() call will raise an exception so any code after the retry will not be reached. This is the Retry exception, it is not handled as an error but rather as a semi-predicate to signify to the worker that the task is to be retried, so that it can store the correct state when a result backend is enabled.



The exc method is used to pass exception information that is used in logs, and when storing task results. Both the exception and the traceback will be available in the task state (if a result backend is enabled).

If the task has a max_retries value the current exception will be re-raised if the max number of retries has been exceeded, but this will not happen if:

  • An exc argument was not given.

    In this case the MaxRetriesExceeded exception will be raised.

  • There is no current exception



    will raise the exc argument given.

Using a custom retry delay



def add(self, x, y):
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)  # override the default and
                                                 # retry in 1 minute

List of Options




The name the task is registered as.

You can set this name manually, or a name will be automatically generated using the module and class name. See Names.



See Context.


Abstract classes are not registered, but are used as the base class for new task types.


The maximum number of attempted retries before giving up. If the number of retries exceeds this value a MaxRetriesExceeded exception will be raised. NOTE: You have to call retry() manually, as it will not automatically retry on exception..

The default value is 3. A value of None will disable the retry limit and the task will retry forever until it succeeds.


Optional tuple of expected error classes that should not be regarded as an actual error.

Errors in this list will be reported as a failure to the result backend, but the worker will not log the event as an error, and no traceback will be included.


@task(throws=(KeyError, HttpNotFound)):
def get_foo():

Error types:

  • Expected errors (in Task.throws)

    Logged with severity INFO, traceback excluded.

  • Unexpected errors

    Logged with severity ERROR, with traceback included.


By default the task will keep track of subtasks called (task.request.children), and this will be stored with the final result in the result backend, available to the client via AsyncResult.children.

This list of task can grow quite big for tasks starting many subtasks, and you can set this attribute to False to disable it.


Default time in seconds before a retry of the task should be executed. Can be either int or float. Default is a 3 minute delay.


Set the rate limit for this task type which limits the number of tasks that can be run in a given time frame. Tasks will still complete when a rate limit is in effect, but it may take some time before it’s allowed to start.




Default is the CELERY_DEFAULT_RATE_LIMIT setting, which if not specified means rate limiting for tasks is disabled by default.

Note that this is a per worker instance rate limit, and not a global rate limit. To enforce a global rate limit (e.g. for an API with a maximum number of requests per second), you must restrict to a given queue.


The hard time limit, in seconds, for this task. If not set then the workers default will be used.


The soft time limit for this task. If not set then the workers default will be used.






Send an email whenever a task of this type fails. Defaults to the CELERY_SEND_TASK_ERROR_EMAILS setting. See Error E-Mails for more information.


If the sending of error emails is enabled for this task, then this is the class defining the logic to send error mails.


A string identifying the default serialization method to use. Defaults to the CELERY_TASK_SERIALIZER setting. Can be pickle, json, yaml, or any custom serialization methods that have been registered with kombu.serialization.registry.

Please see Serializers for more information.


A string identifying the default compression scheme to use.

Defaults to the CELERY_MESSAGE_COMPRESSION setting. Can be gzip, or bzip2, or any custom compression schemes that have been registered with the kombu.compression registry.

Please see Compression for more information.


The result store backend to use for this task. An instance of one of the backend classes in celery.backends. Defaults to app.backend which is defined by the CELERY_RESULT_BACKEND setting.



Note that this means the task may be executed twice if the worker crashes in the middle of execution, which may be acceptable for some applications.

The global default can be overridden by the CELERY_ACKS_LATE setting.



The host name and process id of the worker executing the task will be available in the state metadata (e.g.[‘pid’])

The global default can be overridden by the CELERY_TRACK_STARTED setting.

See also

The API reference for Task.




During its lifetime a task will transition through several possible states, and each state may have arbitrary metadata attached to it. When a task moves into a new state the previous state is forgotten about, but some transitions can be deducted, (e.g. a task now in the FAILED state, is implied to have been in the STARTED state at some point).



You can also define Custom states.

Result Backends

If you want to keep track of tasks or need the return values, then Celery must store or send the states somewhere so that they can be retrieved later. There are several built-in result backends to choose from: SQLAlchemy/Django ORM, Memcached, RabbitMQ/QPid (rpc), MongoDB, and Redis – or you can define your own.


RPC Result Backend (RabbitMQ/QPid)

The RPC result backend (rpc://) is special as it does not actually store the states, but rather sends them as messages. This is an important difference as it means that a result can only be retrieved once, and only by the client that initiated the task. Two different processes can not wait for the same result.

Even with that limitation, it is an excellent choice if you need to receive state changes in real-time. Using messaging means the client does not have to poll for new states.


Database Result Backend


  • Polling the database for new states is expensive, and so you should increase the polling intervals of operations such as result.get().

  • Some databases use a default transaction isolation level that is not suitable for polling tables for changes.

    In MySQL the default transaction isolation level is REPEATABLE-READ, which means the transaction will not see changes by other transactions until the transaction is committed. It is recommended that you change to the READ-COMMITTED isolation level.

Built-in States


Task is waiting for execution or unknown. Any task id that is not known is implied to be in the pending state.



metadata:pid and hostname of the worker process executing the task.


Task has been successfully executed.

metadata:result contains the return value of the task.


Task execution resulted in failure.



Task is being retried.



Task has been revoked.


Custom states

You can easily define your own states, all you need is a unique name. The name of the state is usually an uppercase string. As an example you could have a look at abortable tasks which defines its own custom ABORTED state.

Use update_state() to update a task’s state:

def upload_files(self, filenames):
    for i, file in enumerate(filenames):
        if not self.request.called_directly:

Here I created the state “PROGRESS”, which tells any application aware of this state that the task is currently in progress, and also where it is in the process by having current and total counts as part of the state metadata. This can then be used to create e.g. progress bars.

Creating pickleable exceptions


Tasks that raise exceptions that are not pickleable will not work properly when Pickle is used as the serializer.



# OK:
class HttpError(Exception):

# BAD:
class HttpError(Exception):

    def __init__(self, status_code):
        self.status_code = status_code

# OK:
class HttpError(Exception):

    def __init__(self, status_code):
        self.status_code = status_code



class HttpError(Exception):

        self.status_code = status_code
        self.headers = headers
        self.body = body



The worker wraps the task in a tracing function which records the final state of the task. There are a number of exceptions that can be used to signal this function to change how it treats the return of the task.




Example keeping revoked tasks in a Redis set:

from celery.exceptions import Ignore

def some_task(self):
    if redis.ismember('tasks.revoked',
        raise Ignore()

Example that stores results manually:

from celery import states
from celery.exceptions import Ignore

def get_tweets(self, user):
    timeline = twitter.get_timeline(user)
    if not self.request.called_directly:
        self.update_state(state=states.SUCCESS, meta=timeline)
    raise Ignore()


The task may raise Reject to reject the task message using AMQPs basic_reject method. This will not have any effect unless Task.acks_late is enabled.


Reject can also be used to requeue messages, but please be very careful when using this as it can easily result in an infinite message loop.


import errno
from celery.exceptions import Reject

@app.task(bind=True, acks_late=True)
def render_scene(self, path):
    file = get_file(path)

    # if the file is too big to fit in memory
    # and we can manually inspect the situation.
    except MemoryError as exc:
        raise Reject(exc, requeue=False)
    except OSError as exc:
        if exc.errno == errno.ENOMEM:
            raise Reject(exc, requeue=False)

    # For any other error we retry after 10 seconds.
    except Exception as exc:
        raise self.retry(exc, countdown=10)

Example requeuing the message:

from celery.exceptions import Reject

@app.task(bind=True, acks_late=True)
def requeues(self):
    if not self.request.delivery_info['redelivered']:
        raise Reject('no reason', requeue=True)
    print('received two times')

Consult your broker documentation for more details about the basic_reject method.



Custom task classes


As an example, the following code,

def add(x, y):
    return x + y

will do roughly this behind the scenes:

class _AddTask(app.Task):

    def run(self, x, y):
        return x + y
add = app.tasks[]




If you have a task,

from celery import Task

class NaiveAuthenticateServer(Task):

    def __init__(self):
        self.users = {'george': 'password'}

    def run(self, username, password):
            return self.users[username] == password
        except KeyError:
            return False


This can also be useful to cache resources, e.g. a base Task class that caches a database connection:

from celery import Task

class DatabaseTask(Task):
    abstract = True
    _db = None

    def db(self):
        if self._db is None:
            self._db = Database.connect()
        return self._db

that can be added to tasks like this:

def process_rows():
    for row in process_rows.db.table.all():


Abstract classes


from celery import Task

class DebugTask(Task):
    abstract = True

        print('Task returned: {0!r}'.format(self.request))

def add(x, y):
    return x + y



Handler called after the task returns.

  • status – Current task state.
  • retval – Task return value/exception.
  • task_id – Unique id of the task.
  • args – Original arguments for the task that returned.
  • kwargs – Original keyword arguments for the task that returned.
  • einfoExceptionInfo instance, containing the traceback (if any).

The return value of this handler is ignored.


This is run by the worker when the task fails.

  • exc – The exception raised by the task.
  • task_id – Unique id of the failed task.
  • args – Original arguments for the task that failed.
  • kwargs – Original keyword arguments for the task that failed.
  • einfoExceptionInfo instance, containing the traceback.

The return value of this handler is ignored.



  • exc – The exception sent to retry().
  • task_id – Unique id of the retried task.
  • args – Original arguments for the retried task.
  • kwargs – Original keyword arguments for the retried task.
  • einfoExceptionInfo instance, containing the traceback.

The return value of this handler is ignored.

on_success(self, retval, task_id, args, kwargs)

Run by the worker if the task executes successfully.

  • retval – The return value of the task.
  • task_id – Unique id of the executed task.
  • args – Original arguments for the executed task.
  • kwargs – Original keyword arguments for the executed task.

The return value of this handler is ignored.

How it works



>>> from proj.celery import app
>>> app.tasks
    <@task: celery.chord_unlock>,
    <@task: celery.backend_cleanup>,
    <@task: celery.chord>}

This is the list of tasks built-in to celery. Note that tasks will only be registered when the module they are defined in is imported.

The default loader imports any modules listed in the CELERY_IMPORTS setting.

The entity responsible for registering your task in the registry is the metaclass: TaskType.

If you want to register your task manually you can mark the task as abstract:

class MyTask(Task):
    abstract = True

This way the task won’t be registered, but any task inheriting from it will be.


This means that your workers should always be updated with the same software as the client. This is a drawback, but the alternative is a technical challenge that has yet to be solved.

Tips and Best Practices

Ignore results you don’t want


def mytask(…):

Results can even be disabled globally using the CELERY_IGNORE_RESULT setting.

Disable rate limits if they’re not used

Disabling rate limits altogether is recommended if you don’t have any tasks using them. This is because the rate limit subsystem introduces quite a lot of complexity.

Set the CELERY_DISABLE_RATE_LIMITS setting to globally disable rate limits:


You find additional optimization tips in the Optimizing Guide.

Avoid launching synchronous subtasks


Make your design asynchronous instead, for example by using callbacks.


def update_page_info(url):
    page = fetch_page.delay(url).get()
    info = parse_page.delay(url, page).get()
    store_page_info.delay(url, info)

def fetch_page(url):
    return myhttplib.get(url)

def parse_page(url, page):
    return myparser.parse_document(page)

def store_page_info(url, info):
    return PageInfo.objects.create(url, info)


def update_page_info(url):
    # fetch_page -> parse_page -> store_page

def fetch_page(url):
    return myhttplib.get(url)

def parse_page(page):
    return myparser.parse_document(page)

def store_page_info(info, url):
    PageInfo.objects.create(url=url, info=info)

Here I instead created a chain of tasks by linking together different subtask()‘s. You can read about chains and other powerful constructs at Canvas: Designing Workflows.

Performance and Strategies




However, executing a task does have overhead. A message needs to be sent, data may not be local, etc. So if the tasks are too fine-grained the additional overhead may not be worth it in the end.

See also



Data locality




See also



Since celery is a distributed system, you can’t know in which process, or on what machine the task will be executed. You can’t even know if the task will run in a timely manner.




class Article(models.Model):
    title = models.CharField()
    body = models.TextField()

def expand_abbreviations(article):
    article.body.replace('MyCorp', 'My Corporation')


>>> article = Article.objects.get(id=102)
>>> expand_abbreviations.delay(article)



def expand_abbreviations(article_id):
    article = Article.objects.get(id=article_id)
    article.body.replace('MyCorp', 'My Corporation')

>>> expand_abbreviations(article_id)


Database transactions

Let’s have a look at another example:

from django.db import transaction

def create_article(request):
    article = Article.objects.create(…)

This is a Django view creating an article object in the database, then passing the primary key to a task. It uses the commit_on_success decorator, which will commit the transaction when the view returns, or roll back if the view raises an exception.

There is a race condition if the task starts executing before the transaction has been committed; The database object does not exist yet!

The solution is to always commit transactions before sending tasks depending on state from the current transaction:

def create_article(request):
        article = Article.objects.create(…)


Django 1.6 (and later) now enables autocommit mode by default, and commit_on_success/commit_manually are deprecated.

This means each SQL query is wrapped and executed in individual transactions, making it less likely to experience the problem described above.

However, enabling ATOMIC_REQUESTS on the database connection will bring back the transaction-per-request model and the race condition along with it. In this case, the simple solution is using the @transaction.non_atomic_requests decorator to go back to autocommit for that view only.





The comment model looks like this:

from django.db import models
from django.utils.translation import ugettext_lazy as _

class Comment(models.Model):
    email_address = models.EmailField(_('email address'))
    homepage = models.URLField(_('home page'),
                               blank=True, verify_exists=False)
    comment = models.TextField(_('comment'))
    pub_date = models.DateTimeField(_('Published date'),
                                    editable=False, auto_add_now=True)
    is_spam = models.BooleanField(_('spam?'),
                                  default=False, editable=False)

    class Meta:
        verbose_name = _('comment')
        verbose_name_plural = _('comments')



from django import forms
from django.http import HttpResponseRedirect
from django.template.context import RequestContext
from django.shortcuts import get_object_or_404, render_to_response

from blog import tasks
from blog.models import Comment

class CommentForm(forms.ModelForm):

    class Meta:
        model = Comment

    post = get_object_or_404(Entry, slug=slug)
    remote_addr = request.META.get('REMOTE_ADDR')

    if request.method == 'post':
        form = CommentForm(request.POST, request.FILES)
        if form.is_valid():
            comment =
            # Check spam asynchronously.
            return HttpResponseRedirect(post.get_absolute_url())
        form = CommentForm()

    return render_to_response(template_name, context_instance=context)

To filter spam in comments I use Akismet, the service used to filter spam in comments posted to the free weblog platform Wordpress. Akismet is free for personal use, but for commercial use you need to pay. You have to sign up to their service to get an API key.



from celery import Celery

from akismet import Akismet

from django.core.exceptions import ImproperlyConfigured
from django.contrib.sites.models import Site

from blog.models import Comment


def spam_filter(comment_id, remote_addr=None):

    comment = Comment.objects.get(pk=comment_id)
    current_domain = Site.objects.get_current().domain
    akismet = Akismet(settings.AKISMET_KEY,
    if not akismet.verify_key():
        raise ImproperlyConfigured('Invalid AKISMET_KEY')

    is_spam = akismet.comment_check(user_ip=remote_addr,
    if is_spam:
        comment.is_spam = True

    return is_spam