django-tasks-redis

A Redis/Valkey-backed task queue backend for Django 6.0's built-in task framework.

Features

• Full integration with Django 6.0's task framework (django.tasks)
• Redis Streams for reliable task queuing with consumer groups
• Support for both Redis and Valkey backends
• Delayed task execution with scheduled times
• Priority-based task processing
• Crash recovery with automatic task reclaim
• Django Admin integration for task monitoring and management
• HTTP endpoints for external triggers (webhooks, Cloud Scheduler, etc.)

Architecture

sequenceDiagram
    participant App as Application
    participant Backend as RedisTaskBackend
    participant Redis as Redis/Valkey
    participant Worker as Worker Process

    Note over App,Worker: Task Enqueue
    App->>Backend: task.enqueue(args, kwargs)
    Backend->>Backend: Validate & serialize args
    Note over Backend,Redis: One transaction
    Backend->>Redis: HSET task data (status=READY)
    Backend->>Redis: SADD to results index
    alt run_after in the future
        Backend->>Redis: ZADD to delayed set
    else Ready to run
        Backend->>Redis: XADD to priority stream
    end
    Backend-->>App: TaskResult (id, status=READY)

    Note over App,Worker: Task Execution
    Worker->>Redis: Promote due delayed tasks<br/>(ZREM + XADD in one script)
    Worker->>Redis: XREADGROUP (consumer group)<br/>(own pending messages, then new ones)
    Redis-->>Worker: Message with task_id
    Worker->>Redis: HGET task data
    Redis-->>Worker: Task data
    Worker->>Redis: HSET status=RUNNING
    Worker->>Worker: Execute task function
    alt Success
        Worker->>Redis: HSET status=SUCCESSFUL,<br/>return_value, finished_at
    else Failure
        Worker->>Redis: HSET status=FAILED,<br/>errors, finished_at
    end
    Worker->>Redis: XACK + XDEL (acknowledge and delete the entry)

    Note over App,Worker: Crash Recovery
    Worker->>Redis: XPENDING + XCLAIM stale messages<br/>(claim_timeout exceeded)
    Redis-->>Worker: Messages reassigned to this consumer
    Worker->>Worker: Re-execute tasks<br/>(up to REDIS_MAX_DELIVERIES)

    Note over App,Worker: Result Retrieval (Optional)
    App->>Backend: backend.get_result(task_id)
    Backend->>Redis: HGETALL task data
    Redis-->>Backend: Task data
    Backend-->>App: TaskResult (status, return_value, errors)

Requirements

• Python 3.12+
• Django 6.0+
• Redis 5.0+ or Valkey 7.2+

Installation

pip install django-tasks-redis

Quick Start

1. Add django_tasks_redis to your INSTALLED_APPS:

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

2. Configure the task backend in your Django settings:

TASKS = {
    "default": {
        "BACKEND": "django_tasks_redis.RedisTaskBackend",
        "QUEUES": [],  # Empty list = allow all queue names
        "OPTIONS": {
            "REDIS_URL": "redis://localhost:6379/0",
        },
    },
}

Note: QUEUES controls which queue names are allowed. If omitted, only "default" queue is allowed. Set QUEUES: [] (empty list) to allow all queue names, or specify explicit names like ["default", "emails"].

3. Define a task:

from django.tasks import task


@task
def send_email(to: str, subject: str, body: str):
    # Send email logic here
    pass

4. Enqueue the task:

result = send_email.enqueue("user@example.com", "Hello", "World")
print(f"Task ID: {result.id}")

5. Run the worker:

python manage.py run_redis_tasks

Configuration Options

TASKS = {
    "default": {
        "BACKEND": "django_tasks_redis.RedisTaskBackend",
        "QUEUES": [],  # Empty list = allow all queue names
        "OPTIONS": {
            # Connection settings (use URL or individual settings)
            "REDIS_URL": "redis://localhost:6379/0",
            # Or use individual settings:
            # "REDIS_HOST": "localhost",
            # "REDIS_PORT": 6379,
            # "REDIS_DB": 0,
            # "REDIS_PASSWORD": None,
            # "REDIS_SSL": False,
            # "REDIS_SSL_CA_CERTS": "/path/to/ca.pem",  # CA cert path for TLS (self-signed CA). Requires REDIS_SSL=True (or a rediss:// URL).
            # Connection robustness (passed to redis-py)
            "REDIS_SOCKET_CONNECT_TIMEOUT": 5,  # Seconds to wait for a connection
            "REDIS_HEALTH_CHECK_INTERVAL": 30,  # Seconds before an idle connection is pinged
            # "REDIS_SOCKET_TIMEOUT": None,  # Seconds; must exceed REDIS_BLOCK_TIMEOUT (see below)
            # "REDIS_SOCKET_KEEPALIVE": None,  # Enable TCP keepalive
            # Behavior settings
            "REDIS_RESULT_TTL": 2592000,  # Result retention period (seconds), default 30 days
            "REDIS_COMPLETED_TASK_TTL": 2592000,  # Retention once finished, defaults to REDIS_RESULT_TTL
            "REDIS_KEY_PREFIX": "django_tasks",  # Redis key prefix
            "REDIS_CONSUMER_GROUP": "django_tasks_workers",  # Consumer group name
            "REDIS_CLAIM_TIMEOUT": 300,  # Stale message claim timeout (seconds)
            "REDIS_BLOCK_TIMEOUT": 5000,  # XREADGROUP block timeout (milliseconds)
            "REDIS_MAX_DELIVERIES": 5,  # Give up on a task started this many times without finishing (0 = never)
            "REDIS_SCAN_BATCH_SIZE": 500,  # Tasks read per round trip when walking the index
        },
    },
}

Connection robustness

Without any timeout a worker blocked on a connection Redis no longer answers — a restart, a network partition — waits as long as the kernel allows, and looks idle while it does. REDIS_SOCKET_CONNECT_TIMEOUT bounds establishing a connection and REDIS_HEALTH_CHECK_INTERVAL makes redis-py check an idle one before reusing it.

REDIS_SOCKET_TIMEOUT is left unset on purpose: it also applies to the worker's blocking reads, so a value below REDIS_BLOCK_TIMEOUT makes every fetch raise TimeoutError. Set it above that, in seconds, or leave it alone.

Setting any of these options to None drops the default and lets redis-py apply its own — useful when the value above is wrong for your environment.

Note that redis-py's default Retry multiplies the connect timeout. With REDIS_SOCKET_CONNECT_TIMEOUT=2 against an unroutable host, the default retry policy in redis-py 8.x stretches a 2.0 s wait to roughly 25 s, so the actual wait can be much longer than the configured value.

Delivery guarantees

Tasks are delivered at least once. A worker that dies leaves its message pending; another worker reclaims it after REDIS_CLAIM_TIMEOUT and runs it again. Two consequences are worth planning for:

• REDIS_CLAIM_TIMEOUT must be longer than your longest task. A task still running after the timeout looks exactly like a dead worker at the queue level, and will be reclaimed and executed a second time.
• Task functions should be idempotent. A worker can die between finishing the work and recording the result, in which case the task runs again.

A task that has been started REDIS_MAX_DELIVERIES times without finishing is given up on the next time its message is reclaimed: it is marked FAILED with a TaskAbandoned error, so it shows up in the admin instead of being retried forever. Only starts count, not deliveries, so a message a worker held without running it does not use up an attempt. A task that already finished keeps its result. Set it to 0 to disable the cap.

Management Commands

run_redis_tasks

Start a worker to process tasks:

python manage.py run_redis_tasks [options]

Options:
  --queue QUEUE_NAME      Process only tasks from specific queue
  --backend BACKEND_NAME  Backend name (default: default)
  --continuous            Continuous mode (don't exit)
  --interval SECONDS      Polling interval (default: 1)
  --max-tasks N           Maximum tasks to process (0=unlimited)
  --claim-interval SECS   Stale task claim interval (default: 60)

A worker handles one task at a time. Run several processes to process more, each gets its own consumer in the group.

In --continuous mode the worker waits on the streams for up to REDIS_BLOCK_TIMEOUT instead of polling, so --interval only applies when that wait is disabled (REDIS_BLOCK_TIMEOUT: 0). The block timeout also bounds how long a shutdown signal can take to be noticed.

purge_completed_redis_tasks

Delete completed tasks:

python manage.py purge_completed_redis_tasks [options]

Options:
  --days N                Delete tasks completed N+ days ago
  --status STATUS         Target status (default: SUCCESSFUL,FAILED)
  --batch-size N          Tasks read per round trip (default: REDIS_SCAN_BATCH_SIZE)
  --dry-run               Only show count, don't delete
  --backend BACKEND_NAME  Backend name (default: default)

Django Admin

The package provides Django Admin integration for viewing and managing tasks:

• View task list with status, priority, queue
• Search a task by id
• Run selected tasks (requires run_redistask)
• Retry failed tasks (requires run_redistask)
• Delete tasks (requires delete_redistask)

The admin reads the default backend.

Permissions

Tasks live in Redis, so RedisTask is an unmanaged model with no table. It still takes a migrate run for its permissions to be created, after which they are granted like any other model's:

Tasks cannot be added or edited through the admin, so no add or change permission exists.

HTTP Endpoints

Include the URLs in your project:

from django.urls import include, path

urlpatterns = [
    # ...
    path("tasks/", include("django_tasks_redis.urls")),
]

Available endpoints:

• POST /tasks/run/ - Process multiple tasks
• POST /tasks/run-one/ - Process a single task
• POST /tasks/execute/<task_id>/ - Execute specific task by ID
• GET /tasks/status/<task_id>/ - Get task status
• POST /tasks/purge/ - Purge completed tasks

These endpoints run tasks, expose their arguments and results, and delete task history, so they answer 403 until the backend says how to authenticate them. Override get_auth_handler() to open them. The handler returns None to let the request through, or a response to refuse it:

from django.conf import settings
from django.http import JsonResponse

from django_tasks_redis.backends import RedisTaskBackend


class MyTaskBackend(RedisTaskBackend):
    def get_auth_handler(self):
        def handler(request):
            if request.headers.get("X-Task-Token") != settings.TASK_ENDPOINT_TOKEN:
                return JsonResponse({"error": "Forbidden"}, status=403)
            return None

        return handler

Then point BACKEND at myapp.backends.MyTaskBackend. The endpoints are csrf_exempt, so the handler is the only thing standing between the caller and task execution: authenticate on something the caller has to prove, not on anything the request can claim about itself.

POST /tasks/run/ drains the whole queue in the request by default; pass max_tasks to bound it.

Public API

The executor module provides functions for programmatic task management:

from django_tasks_redis import executor

# Process tasks
result = executor.process_one_task(queue_name="default")
results = executor.process_tasks(max_tasks=10)

# Execute specific task
result = executor.run_task_by_id(task_id, allow_retry=True)

# Get pending task count
count = executor.get_pending_task_count()

# Purge completed tasks
deleted = executor.purge_completed_tasks(days=7)

Contributing

Issues and pull requests are welcome. See CONTRIBUTING.md for how to set up a development environment and what a pull request needs.

License

MIT License