DOCSEARCHbyAlgolia
Submit Docs

django-tasks Documentation

repository·master·Indexed 21 days ago

https://github.com/realorangeone/django-tasks

A backport of Django's built-in Tasks framework (version 0.12.0) that allows developers to define, enqueue, and manage asynchronous tasks within a Django application. It provides a @task decorator for registering functions, a system for configuring multiple task backends via the TASKS setting, and tools for inspecting task outcomes using TaskResult and TaskResultStatus. The library includes built-in backends such as ImmediateBackend and DummyBackend, and supports task lifecycle monitoring through Django signals.

Tokens
5.1K
Snippets
24
Records
28
Agent score
71%

What's inside django-tasks

  1. Install and configure django-tasks

    master

    Install the package via pip:

    python -m pip install django-tasks

    Add django_tasks to your INSTALLED_APPS in settings.py:

    INSTALLED_APPS = [
    # ...
    "django_tasks",
]

    Configure a backend in your TASKS setting. If omitted, it defaults to ImmediateBackend. Available built-in backends include:

    • django_tasks.backends.immediate.ImmediateBackend: Executes the task immediately in the current thread.
    • django_tasks.backends.dummy.DummyBackend: Does not execute tasks, only stores them (useful for testing).

    Note: Prior to version 0.12.0, database and RQ backends were included via separate packages (django-tasks-db and django-tasks-rq).

    INSTALLED_APPS = [
    # ...
    "django_tasks",
]

TASKS = {
    "default": {
        "BACKEND": "django_tasks.backends.immediate.ImmediateBackend"
    }
}

  2. Configure allowed queue names

    master

    By default, tasks are enqueued to the "default" queue. You can restrict tasks to specific queues by defining QUEUES in your TASKS configuration. Enqueueing to an unknown queue name will raise an InvalidTask error.

    To disable queue name validation, set QUEUES to an empty list [].

    TASKS = {
    "default": {
        "BACKEND": "django_tasks.backends.immediate.ImmediateBackend",
        "QUEUES": ["default", "special"]
    }
}

  3. Define a task using the @task decorator

    master

    Use the @task decorator to transform a standard Python function into a Task. You can configure the task's priority, queue name, backend, and whether it should receive a TaskContext during execution.

    Configuration Options

    • priority: An integer (default 0). Range is typically -100 to 100.
    • queue_name: The name of the queue to run on (default "default").
    • backend: The name of the backend to use (default "default").
    • takes_context: If True, the decorated function must accept a TaskContext as its first argument.

    Usage Patterns

    Standard Task:

    @task
def my_function(x: int, y: int) -> int:
    return x + y

    Task with Arguments:

    @task(priority=10, queue_name="high_priority")
def heavy_computation(data: dict):
    ...

    Task with Context: If takes_context=True, the function signature must include TaskContext.

    @task(takes_context=True)
def task_with_context(context: TaskContext, name: str):
    print(f"Attempt number: {context.attempt}")

  5. Inspect TaskResult status and values

    master

    The TaskResult object provides several attributes to inspect the outcome of a task:

    • status: The current state (e.g., TaskResultStatus.SUCCESSFUL, TaskResultStatus.READY).
    • return_value: The value returned by the task. Accessing this on a non-successful task raises a ValueError.
    • errors: A list containing error information if the task failed. Currently, it contains a single element with exception_class and a traceback (as a string).
    • attempts: The number of times the task has run (currently 0 or 1).
    • last_attempted_at: The timestamp of the last attempt.

    If a result was updated in the background, call .refresh() to update the local object's values.

    # Checking success and value
if result.status == TaskResultStatus.SUCCESSFUL:
    print(result.return_value)

# Refreshing data from the backend
result.refresh()

# Inspecting errors
if result.errors:
    error = result.errors[0]
    print(f"Error: {error.exception_class}")
    print(f"Traceback: {error.traceback}")

  6. Retrieve task results by ID

    master

    If you need to retrieve a task result from a different part of your application (e.g., a different request or task), use the id from the original TaskResult and call get_result() or aget_result().

    • To retrieve a result for a specific task type: task_function.get_result(result_id)
    • To retrieve a result for ANY task type: default_task_backend.get_result(result_id)

    Note: The result_id is an opaque string (up to 64 characters) and generation is backend-specific.

    # 1. Get the ID from the original result
result_id = result.id

# 2. Retrieve it later elsewhere
# Using the task class:
new_result = calculate_meaning_of_life.get_result(result_id)

# OR using the backend directly (works for any task type):
from django_tasks import default_task_backend
any_result = default_task_backend.get_result(result_id)

  7. Use django-tasks signals

    master

    You can respond to task lifecycle events using Django signals. The sender for these signals is the backend class, and they are provided with the task_result object.

    Available signals:

    • django_tasks.signals.task_enqueued: Triggered when a task is enqueued.
    • django_tasks.signals.task_started: Triggered immediately before a task starts executing.
    • django_tasks.signals.task_finished: Triggered when a task finishes (either SUCCESSFUL or FAILED).

  8. Introspect backend capabilities

    master

    Since django-tasks supports multiple backends, you can check at runtime if a backend supports specific features. This is useful for graceful degradation or system checks.

    Available capability flags:

    • supports_defer: Supports the run_after attribute.
    • supports_async_task: Supports enqueuing coroutines.
    • supports_get_result: Supports retrieving results after the fact from any thread/process.
    • supports_priority: Supports executing tasks in a specific priority order.
    from django_tasks import default_task_backend

if default_task_backend.supports_get_result:
    # Proceed with logic that requires result retrieval
    pass

  9. Enqueue a task

    master

    To execute a task, call the .enqueue() method on the decorated function. You can pass arguments to the task directly into enqueue().

    Calling .enqueue() returns a TaskResult object, which can be used to check the status and retrieve the return value.

    # Enqueueing a task with no arguments
result = calculate_meaning_of_life.enqueue()

# Enqueueing a task with arguments
# result = my_task.enqueue(arg1, arg2)

  10. Define a task with the @task decorator

    master

    Use the @task() decorator to turn a function into a task. You can customize task behavior using the .using() method on the decorated function.

    Supported customization options:

    • priority: Integer between -100 and 100 (default: 0). Higher numbers are higher priority.
    • queue_name: The name of the queue to run the task on.
    • backend: The name of the backend to use (as defined in TASKS).
    • run_after: A specific time for the task to run (requires backend support for supports_defer).

    To pass context (like attempt count) to your task, set takes_context=True in the decorator and ensure the task function accepts context: TaskContext as its first argument.

    from django_tasks import task, TaskContext

@task(takes_context=True)
def calculate_meaning_of_life(context: TaskContext) -> int:
    # context.attempt provides the current attempt number
    return 42

# Customize task behavior at call-site
modified_task = calculate_meaning_of_life.using(priority=10)

  11. Configure task backends in Django settings

    master

    Task backends are managed via the TASKS setting in your Django configuration. The TaskBackendHandler looks for a dictionary under TASKS where keys are aliases and values are configuration dictionaries containing a BACKEND key (the import path to the backend class).

    If no settings are provided, the system defaults to using django_tasks.backends.immediate.ImmediateBackend via the DEFAULT_TASK_BACKEND_ALIAS.

    # Example Django settings.py configuration
TASKS = {
    'default': {
        'BACKEND': 'django_tasks.backends.immediate.ImmediateBackend',
    },
    'redis_queue': {
        'BACKEND': 'path.to.your.CustomRedisBackend',
        'connection_url': 'redis://localhost:6379/0',
    },
}

  12. Refresh a cached TaskResult

    master

    Because TaskResult objects are often snapshots of data from a backend, you may need to reload the latest data from the store to see updated statuses or return values.

    • refresh(): Synchronously reloads the task data.
    • arefresh(): Asynchronously reloads the task data.
    result = my_task.get_result(result_id)

# ... later ...

result.refresh()
if result.is_finished:
    print(result.return_value)