Mastering Celery Tasks: Instantiation, Naming, Binding, Retries, and Context

Task Fundamentals

Continuing from previous articles, we explore Celery Tasks in depth. A Task is the core of Celery, represented by the prototype class celery.app.task:Task, which provides two essential capabilities: sending a task message to a queue and declaring the function that a worker will execute.

Task Instantiation

Decorate a regular function with @app.task to create a task function.

from proj.celery import app

@app.task
def add(x, y):
    return x + y

Note: the task function becomes an instance of celery.app.task:Task , not a plain function.

>> from proj.task.tasks import add
>>> dir(add)
['AsyncResult', 'MaxRetriesExceededError', ..., 'apply_async', 'delay', 'name', 'on_bound', 'on_failure', 'on_retry', 'on_success', 'request', 'retry', 'subtask', ...]

Therefore you can call add.delay or add.apply_async, which are attributes of the Task instance.

>> add.apply_async
<bound method add.apply_async of <@task: proj.task.tasks.add at 0x7fedb363a790>>
>>> add.delay
<bound method add.delay of <@task: proj.task.tasks.add at 0x7fedb363a790>>

The @app.task decorator essentially performs Task instantiation, and its parameters become the Task's initialization arguments.

Decorator order matters: @app.task should be placed last when multiple decorators are used.

app.task
@decorator2
@decorator1
def add(x, y):
    return x + y

Task Naming

Each task has a unique name used by workers to locate the function. By default, Celery auto‑generates the name from the function's full import path.

>> add.name
u'proj.task.tasks.add'

You can explicitly set a name via the decorator.

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

>>> from proj.task.tasks import add
>>> add.name
'new_name'

Custom names are discouraged unless you fully understand the implications.

Task Binding

Because a task function is a Task instance, it can use the self binding feature.

@app.task(bind=True)
def add(self, x, y):
    print("self: ", self)
    return x + y

>>> add.delay(1, 2)
<AsyncResult: 1982dc85-694b-4ceb-849b-5f69e40b4fe9>

Binding provides access to many advanced Task features, such as retries and request context.

Task Retries

Retrying a task is done via Task.retry, which re‑queues the task.

@app.task(bind=True, max_retries=3)
def send_twitter_status(self, oauth, tweet):
    try:
        twitter = Twitter(oauth)
        twitter.update_status(tweet)
    except (Twitter.FailWhaleError, Twitter.LoginError) as exc:
        raise self.retry(exc=exc)

max_retries sets the maximum number of retry attempts.

exc passes the exception information to the log (requires a result backend).

You can configure automatic retries for specific exceptions.

# Retry only on FailWhaleError, up to 5 times.
@app.task(autoretry_for=(FailWhaleError,), retry_kwargs={'max_retries': 5})
def refresh_timeline(user):
    return twitter.refresh_timeline(user)

Task Request Context

When a worker executes a task, a request context is provided so the task can access its own metadata and state.

@app.task(bind=True)
def dump_context(self, x, y):
    print('Executing task id {0.id}, args: {0.args!r} kwargs: {0.kwargs!r}'.format(self.request))

>>> from proj.task.tasks import dump_context
>>> dump_context.delay(1, 2)
<AsyncResult: 00bc9f96-98df-4bca-a4a3-4774c535a44c>

Inspecting the request context helps diagnose task execution status. See the official Celery documentation for the full list of request attributes.

Task Inheritance

By default, @app.task creates an instance of the native Task class, which covers most cases. For more complex scenarios, you can subclass Task to create specialized base classes.

1. Replace the default base class for all tasks:

def make_app(context):
    app = Celery('proj')
    app.config_from_object('proj.celeryconfig')
    # ... configure queues, beat schedule, etc.
    TaskBase = app.Task
    class ContextTask(TaskBase):
        abstract = True
        context = ctx
        def __call__(self, *args, **kwargs):
            LOG.info("Invoked celery task starting: %(name)s[%(id)s]", {'name': self.name, 'id': self.request.id})
            return super(ContextTask, self).__call__(*args, **kwargs)
        def on_success(self, retval, task_id, args, kwargs):
            LOG.info("Task %(id)s success: [%(ret)s].", {'id': task_id, 'ret': retval})
            return super(ContextTask, self).on_success(retval, task_id, args, kwargs)
        def on_failure(self, exc, task_id, args, kwargs, einfo):
            msg = "Task [%(id)s] failed:
args : %(args)s
kwargs : %(kw)s
detail :%(err)s" % {
                'id': task_id, 'args': args, 'kw': kwargs, 'err': six.text_type(exc)}
            LOG.exception(msg)
            return super(ContextTask, self).on_failure(exc, task_id, args, kwargs, einfo)
    app.Task = ContextTask
    return app

2. Create different base classes for different task types:

class JSONTask(celery.Task):
    serializer = 'json'
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print('{0!r} failed: {1!r}'.format(task_id, exc))

class XMLTask(celery.Task):
    serializer = 'xml'
    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print('{0!r} failed: {1!r}'.format(task_id, exc))

@task(base=JSONTask)
def add_json(x, y):
    raise KeyError()

@task(base=XMLTask)
def add_xml(x, y):
    raise KeyError()