Metadata-Version: 2.4
Name: django-tasks-redis
Version: 0.2.1
Summary: A Redis/Valkey-backed task queue backend for Django 6.0's built-in task framework
Author-email: Shinya Okano <tokibito@gmail.com>
Maintainer-email: Shinya Okano <tokibito@gmail.com>
License-Expression: MIT
Project-URL: Homepage, https://github.com/tokibito/django-tasks-redis
Project-URL: Repository, https://github.com/tokibito/django-tasks-redis
Project-URL: Issues, https://github.com/tokibito/django-tasks-redis/issues
Keywords: django,tasks,redis,valkey,async,queue,background
Classifier: Development Status :: 3 - Alpha
Classifier: Environment :: Web Environment
Classifier: Framework :: Django
Classifier: Framework :: Django :: 6.0
Classifier: Framework :: Django :: 6.1
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: Django>=6.0
Requires-Dist: redis>=5.0.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-django>=4.5; extra == "dev"
Requires-Dist: ruff>=0.8; extra == "dev"
Dynamic: license-file

# django-tasks-redis

[![CI](https://github.com/tokibito/django-tasks-redis/actions/workflows/ci.yml/badge.svg)](https://github.com/tokibito/django-tasks-redis/actions/workflows/ci.yml)
[![PyPI version](https://badge.fury.io/py/django-tasks-redis.svg)](https://badge.fury.io/py/django-tasks-redis)
[![Python versions](https://img.shields.io/pypi/pyversions/django-tasks-redis.svg)](https://pypi.org/project/django-tasks-redis/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

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

```mermaid
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

```bash
pip install django-tasks-redis
```

## Quick Start

1. Add `django_tasks_redis` to your `INSTALLED_APPS`:

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

2. Configure the task backend in your Django settings:

```python
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:

```python
from django.tasks import task


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

4. Enqueue the task:

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

5. Run the worker:

```bash
python manage.py run_redis_tasks
```

## Configuration Options

```python
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` has no bound by default, on purpose: it also applies to
the worker's blocking reads, so a value at or below `REDIS_BLOCK_TIMEOUT` makes
every idle wait raise `TimeoutError`. Set it above that, in seconds, or leave it
alone; the backend logs a warning at startup when the two do not fit. It is
passed to redis-py as `None` rather than left out, because redis-py 8 otherwise
applies a 5 second default of its own, the same length as the default block.

Setting any of the other 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:

```bash
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:

```bash
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:

| Permission | Grants |
| --- | --- |
| `view_redistask` | Read the task list and a task's detail page |
| `run_redistask` | Run and retry tasks |
| `delete_redistask` | Delete tasks from Redis |

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

## HTTP Endpoints

Include the URLs in your project:

```python
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:

```python
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:

```python
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](CONTRIBUTING.md)
for how to set up a development environment and what a pull request needs.

## License

MIT License
