Task implementation: request context and the task base class.

Module Contents


Context(self,*args,**kwargs) Task request variables (Task.request).
Task() Task base class.


_strflags(flags, default="")
_reprtask(task, fmt=None, flags=None)
class Context(*args, **kwargs)

Task request variables (Task.request).

__init__(*args, **kwargs)
update(*args, **kwargs)
get(key, default=None)
class Task

Task base class.

When called tasks apply the run() method. This method must be defined by all tasks (that is unless the __call__() method is overridden).

Called when the task is bound to an app.

This class method can be defined to do additional actions when the task class is bound to an app.
add_around(attr, around)
__call__(*args, **kwargs)
run(*args, **kwargs)

The body of the task executed by workers.

start_strategy(app, consumer, **kwargs)
delay(*args, **kwargs)

Star argument version of apply_async().

Does not support the extra options enabled by apply_async().

*args (Any): Positional arguments passed on to the task. **kwargs (Any): Keyword arguments passed on to the task.
celery.result.AsyncResult: Future promise.
apply_async(args=None, kwargs=None, task_id=None, producer=None, link=None, link_error=None, shadow=None, **options)

Apply tasks asynchronously by sending a message.


args (Tuple): The positional arguments to pass on to the task.

kwargs (Dict): The keyword arguments to pass on to the task.

countdown (float): Number of seconds into the future that the
task should execute. Defaults to immediate execution.
eta (~datetime.datetime): Absolute time and date of when the task
should be executed. May not be specified if countdown is also supplied.
expires (float, ~datetime.datetime): Datetime or
seconds in the future for the task should expire. The task won’t be executed after the expiration time.
shadow (str): Override task name used in logs/monitoring.
Default is retrieved from shadow_name().
connection (kombu.Connection): Re-use existing broker connection
instead of acquiring one from the connection pool.
retry (bool): If enabled sending of the task message will be
retried in the event of connection loss or failure. Default is taken from the :setting:`task_publish_retry` setting. Note that you need to handle the producer/connection manually for this to work.
retry_policy (Mapping): Override the retry policy used.
See the :setting:`task_publish_retry_policy` setting.
queue (str, kombu.Queue): The queue to route the task to.
This must be a key present in :setting:`task_queues`, or :setting:`task_create_missing_queues` must be enabled. See guide-routing for more information.
exchange (str, kombu.Exchange): Named custom exchange to send the
task to. Usually not used in combination with the queue argument.
routing_key (str): Custom routing key used to route the task to a
worker server. If in combination with a queue argument only used to specify custom routing keys to topic exchanges.
priority (int): The task priority, a number between 0 and 9.
Defaults to the priority attribute.
serializer (str): Serialization method to use.
Can be pickle, json, yaml, msgpack or any custom serialization method that’s been registered with kombu.serialization.registry. Defaults to the serializer attribute.
compression (str): Optional compression method
to use. Can be one of zlib, bzip2, or any custom compression methods registered with kombu.compression.register(). Defaults to the :setting:`task_compression` setting.
link (~@Signature): A single, or a list of tasks signatures
to apply if the task returns successfully.
link_error (~@Signature): A single, or a list of task signatures
to apply if an error occurs while executing the task.
producer (kombu.Producer): custom producer to use when publishing
the task.
add_to_parent (bool): If set to True (default) and the task
is applied while executing another task, then the result will be appended to the parent tasks request.children attribute. Trailing can also be disabled by default using the trail attribute

publisher (kombu.Producer): Deprecated alias to producer.

headers (Dict): Message headers to be included in the message.

celery.result.AsyncResult: Promise of future evaluation.
TypeError: If not enough arguments are passed, or too many
arguments are passed. Note that signature checks may be disabled by specifying @task(typing=False).
kombu.exceptions.OperationalError: If a connection to the
transport cannot be made, or if the connection is lost.
Also supports all keyword arguments supported by kombu.Producer.publish().
shadow_name(args, kwargs, options)

Override for custom task name in worker logs/monitoring.

from celery.utils.imports import qualname

def shadow_name(task, args, kwargs, options):
    return qualname(args[0])

@app.task(shadow_name=shadow_name, serializer='pickle')
def apply_function_async(fun, *args, **kwargs):
    return fun(*args, **kwargs)
args (Tuple): Task positional arguments. kwargs (Dict): Task keyword arguments. options (Dict): Task execution options.
signature_from_request(request=None, args=None, kwargs=None, queue=None, **extra_options)
retry(args=None, kwargs=None, exc=None, throw=True, eta=None, countdown=None, max_retries=None, **options)

Retry the task.

>>> from imaginary_twitter_lib import Twitter
>>> from proj.celery import app
>>> @app.task(bind=True)
... def tweet(self, auth, message):
...     twitter = Twitter(oauth=auth)
...     try:
...         twitter.post_status_update(message)
...     except twitter.FailWhale as exc:
...         # Retry in 5 minutes.
...         raise self.retry(countdown=60 * 5, exc=exc)
Although the task will never return above as retry raises an exception to notify the worker, we use raise in front of the retry to convey that the rest of the block won’t be executed.

args (Tuple): Positional arguments to retry with. kwargs (Dict): Keyword arguments to retry with. exc (Exception): Custom exception to report when the max retry

limit has been exceeded (default: @MaxRetriesExceededError).

If this argument is set and retry is called while an exception was raised (sys.exc_info() is set) it will attempt to re-raise the current exception.

If no exception was raised it will raise the exc argument provided.

countdown (float): Time in seconds to delay the retry for. eta (~datetime.datetime): Explicit time and date to run the

retry at.
max_retries (int): If set, overrides the default retry limit for
this execution. Changes to this parameter don’t propagate to subsequent task retry attempts. A value of None, means “use the default”, so if you want infinite retries you’d have to set the max_retries attribute of the task to None first.

time_limit (int): If set, overrides the default time limit. soft_time_limit (int): If set, overrides the default soft

time limit.
throw (bool): If this is False, don’t raise the
@Retry exception, that tells the worker to mark the task as being retried. Note that this means the task will be marked as failed if the task raises an exception, or successful if it returns after the retry call.

**options (Any): Extra options to pass on to apply_async().

To tell the worker that the task has been re-sent for retry. This always happens, unless the throw keyword argument has been explicitly set to False, and is considered normal operation.
apply(args=None, kwargs=None, link=None, link_error=None, task_id=None, retries=None, throw=None, logfile=None, loglevel=None, headers=None, **options)

Execute this task locally, by blocking until the task returns.


args (Tuple): positional arguments passed on to the task. kwargs (Dict): keyword arguments passed on to the task. throw (bool): Re-raise task exceptions.

Defaults to the :setting:`task_eager_propagates` setting.
celery.result.EagerResult: pre-evaluated result.
AsyncResult(task_id, **kwargs)

Get AsyncResult instance for this kind of task.

task_id (str): Task id to get result for.
signature(args=None, *starargs, **starkwargs)

Create signature.

signature: object for
this task, wrapping arguments and execution options for a single task invocation.
s(*args, **kwargs)

Create signature.

Shortcut for .s(*a, **k) -> .signature(a, k).

si(*args, **kwargs)

Create immutable signature.

Shortcut for .si(*a, **k) -> .signature(a, k, immutable=True).

chunks(it, n)

Create a chunks task for this task.


Create a xmap task from it.


Create a xstarmap task from it.

send_event(type_, retry=True, retry_policy=None, **fields)

Send monitoring event message.

This can be used to add custom event types in :pypi:`Flower` and other monitors.

type_ (str): Type of event, e.g. "task-failed".
Keyword Arguments:
retry (bool): Retry sending the message
if the connection is lost. Default is taken from the :setting:`task_publish_retry` setting.
retry_policy (Mapping): Retry settings. Default is taken
from the :setting:`task_publish_retry_policy` setting.
**fields (Any): Map containing information about the event.
Must be JSON serializable.

Replace this task, with a new task inheriting the task id.

New in version 4.0.

sig (~@Signature): signature to replace with.
~@Ignore: This is always raised, so the best practice is to always use raise self.replace(...) to convey to the reader that the task won’t continue after being replaced.
add_to_chord(sig, lazy=False)

Add signature to the chord the current task is a member of.

New in version 4.0.

Currently only supported by the Redis result backend.


sig (~@Signature): Signature to extend chord with. lazy (bool): If enabled the new task won’t actually be called,

and sig.delay() must be called manually.
update_state(task_id=None, state=None, meta=None)

Update task state.

task_id (str): Id of the task to update.
Defaults to the id of the current task.

state (str): New state. meta (Dict): State meta-data.

on_success(retval, task_id, args, kwargs)

Success handler.

Run by the worker if the task executes successfully.

retval (Any): The return value of the task. task_id (str): Unique id of the executed task. args (Tuple): Original arguments for the executed task. kwargs (Dict): Original keyword arguments for the executed task.
None: The return value of this handler is ignored.
on_retry(exc, task_id, args, kwargs, einfo)

Retry handler.

This is run by the worker when the task is to be retried.

exc (Exception): The exception sent to retry(). task_id (str): Unique id of the retried task. args (Tuple): Original arguments for the retried task. kwargs (Dict): Original keyword arguments for the retried task. einfo (~billiard.einfo.ExceptionInfo): Exception information.
None: The return value of this handler is ignored.
on_failure(exc, task_id, args, kwargs, einfo)

Error handler.

This is run by the worker when the task fails.

exc (Exception): The exception raised by the task. task_id (str): Unique id of the failed task. args (Tuple): Original arguments for the task that failed. kwargs (Dict): Original keyword arguments for the task that failed. einfo (~billiard.einfo.ExceptionInfo): Exception information.
None: The return value of this handler is ignored.
after_return(status, retval, task_id, args, kwargs, einfo)

Handler called after the task returns.

status (str): Current task state. retval (Any): Task return value/exception. task_id (str): Unique id of the task. args (Tuple): Original arguments for the task. kwargs (Dict): Original keyword arguments for the task. einfo (~billiard.einfo.ExceptionInfo): Exception information.
None: The return value of this handler is ignored.
push_request(*args, **kwargs)



Get current request object.