Metadata-Version: 2.4
Name: django-tasks-redis
Version: 0.3.0
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
- Graceful shutdown: workers finish the running task before exiting on `SIGTERM`
- 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: Claim: status READY→RUNNING<br/>(one script, and a lost claim is acknowledged and skipped)
    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)
    Worker->>Redis: XGROUP DELCONSUMER idle consumers<br/>that hold nothing

    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)
  --shutdown-timeout SECS   Maximum wait for the running task after SIGTERM/SIGINT
                            before forcing exit (0=wait indefinitely, default: 0)
  --no-graceful-shutdown    Do not install SIGTERM/SIGINT handlers
  --empty-exit-code CODE    Exit code when no task was processed (default: 0)
  --failed-exit-code CODE   Exit code when a task failed or could not be run
                            (default: 0)
```

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 wait is taken in one
second steps, so a shutdown signal is noticed within about a second whatever
the block timeout. See [Graceful Shutdown](#graceful-shutdown).

### 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)
```

## Graceful Shutdown

When a worker is redeployed, the orchestrator (Kubernetes, Cloud Run, systemd,
Docker, supervisord, ...) sends `SIGTERM` and kills the process with `SIGKILL`
after a grace period. Without any handling, a task that happens to be running
at that moment is killed halfway through, and only runs again once the
stale-message sweep of another worker finds its message after
`REDIS_CLAIM_TIMEOUT`.

`run_redis_tasks` installs `SIGTERM` and `SIGINT` handlers by default
(`SIGBREAK` too on Windows, see [On Windows](#on-windows)):

1. On the first signal the worker stops fetching new tasks. A worker waiting
   on the streams stops waiting within about a second.
2. The task currently being executed keeps running until it finishes and its
   result is written to Redis, and its message is acknowledged.
3. The worker removes its consumer from the group and exits with status
   code 0.

```console
$ python manage.py run_redis_tasks --continuous
Starting Redis task worker: worker-1-3f2a9c11
  Backend: default
  Continuous: True
  Graceful shutdown: enabled (timeout=unlimited)
Processed task 1e2d5c0a: SUCCESSFUL
^C
Received SIGINT: no new tasks will be started. Waiting for the running task to finish (send the signal again to force exit).
Processed task 7b1f09d3: SUCCESSFUL

Shutdown complete (no task was interrupted).
Worker stopped. Processed 2 task(s).
```

### Shutdown timeout

By default the worker waits as long as the running task needs. Use
`--shutdown-timeout` to put an upper bound on it, so the process exits on its
own terms instead of being `SIGKILL`ed by the platform:

```bash
python manage.py run_redis_tasks --continuous --shutdown-timeout 25
```

If the task is still running when the timeout expires, the process exits
immediately with status code 1. The task is not lost: its message stays
pending and its hash `RUNNING`, and another worker reclaims and runs it again
after `REDIS_CLAIM_TIMEOUT`, so this is one of the cases the
[at-least-once guarantee](#delivery-guarantees) covers. Set the timeout to a
value slightly below the platform's termination grace period, and keep the
grace period longer than your longest task whenever possible.

Sending the signal a second time (for example pressing Ctrl-C twice) also
forces an immediate exit, with the same consequences.

### On Windows

On Windows the handlers are installed for `SIGINT`, `SIGTERM` and `SIGBREAK`,
but only two of them are ever delivered:

| What stops the worker | What Python sees | What happens |
|-----------------------|------------------|--------------|
| Ctrl-C in the console; NSSM or WinSW stopping the service | `SIGINT` | Graceful shutdown |
| Ctrl-Break; a supervisor sending `CTRL_BREAK_EVENT` to the worker's process group | `SIGBREAK` | Graceful shutdown |
| `taskkill /F`, `Stop-Process`, Task Scheduler's *End task* | nothing | Hard kill |

Nothing on Windows sends `SIGTERM`: `taskkill /F`, `Stop-Process` and *End
task* are all `TerminateProcess`, which no handler sees. `taskkill` without
`/F` posts `WM_CLOSE` to the process's windows, and a console worker has
none, so it does nothing either. After a hard kill the running task is not
lost: its message stays pending and another worker reclaims and runs it again
after `REDIS_CLAIM_TIMEOUT`, the same route as a crash.

So on Windows, run the worker under something that sends Ctrl-C or Ctrl-Break
to stop it, and give it longer than `--shutdown-timeout` before it gives up
and terminates the process. The [NSSM and WinSW](#deployment-examples) examples
below do; for a Windows service written in-house, start the worker with
`CREATE_NEW_PROCESS_GROUP` and stop it with
`GenerateConsoleCtrlEvent(CTRL_BREAK_EVENT, pid)`. Ctrl-C reaches every
process on the console, Ctrl-Break only the group it is sent to, which is
why a supervisor prefers the latter.

Redis itself has no official Windows build. On a Windows-only host the
production option is [Memurai](https://www.memurai.com/) Enterprise (the
Developer edition prohibits production use); otherwise Redis or Valkey runs on
a Linux host or in WSL and only the worker runs on Windows. The
[redis-windows](https://github.com/redis-windows/redis-windows) build the test
suite uses in CI describes itself as being for local development. The backend
has no server-type-specific code, so Memurai is expected to work, but the
suite has not been run against it.

### Cooperating from inside a task

Long running tasks can check whether a shutdown was requested and stop early,
so the worker does not have to wait for the whole task to complete:

```python
from django.tasks import task

from django_tasks_redis import is_shutdown_requested


@task
def import_rows(row_ids):
    processed = []
    for row_id in row_ids:
        if is_shutdown_requested():
            # Requeue the remaining work and return early
            import_rows.enqueue([i for i in row_ids if i not in processed])
            break
        handle(row_id)
        processed.append(row_id)
    return len(processed)
```

`is_shutdown_requested()` returns `False` when no worker with graceful shutdown
is active, so tasks using it stay safe to call from a web request, a test, or
the HTTP endpoints.

### Deployment examples

**Kubernetes** - set `terminationGracePeriodSeconds` longer than the worker's
shutdown timeout:

```yaml
spec:
  terminationGracePeriodSeconds: 60
  containers:
    - name: worker
      command:
        - python
        - manage.py
        - run_redis_tasks
        - --continuous
        - --shutdown-timeout=50
```

**systemd** - `TimeoutStopSec` controls how long systemd waits before
`SIGKILL`:

```ini
[Service]
ExecStart=/srv/app/venv/bin/python manage.py run_redis_tasks --continuous --shutdown-timeout=50
KillSignal=SIGTERM
TimeoutStopSec=60
Restart=always
```

**Docker / Docker Compose** - `docker stop` sends `SIGTERM` and waits for
`--time` (10 seconds by default):

```yaml
services:
  worker:
    command: python manage.py run_redis_tasks --continuous --shutdown-timeout=25
    stop_grace_period: 30s
```

Make sure the worker is PID 1 or that the signal reaches it (use the exec form
of `CMD`, or an init such as `tini`, rather than wrapping the command in a
shell script that swallows signals).

**Windows service with NSSM** - [NSSM](https://nssm.cc/) stops a service by
sending Ctrl-C, then `WM_CLOSE`, then `WM_QUIT`, then `TerminateProcess`,
waiting `AppStopMethodConsole` milliseconds after the Ctrl-C. The default is
1500, so set it longer than `--shutdown-timeout` or the worker is killed
mid-task anyway:

```bat
nssm install redis-task-worker C:\srv\app\venv\Scripts\python.exe ^
    manage.py run_redis_tasks --continuous --shutdown-timeout=50
nssm set redis-task-worker AppDirectory C:\srv\app
nssm set redis-task-worker AppEnvironmentExtra DJANGO_SETTINGS_MODULE=myproject.settings
nssm set redis-task-worker AppStopMethodConsole 60000
nssm set redis-task-worker AppStdout C:\srv\app\logs\worker.log
nssm set redis-task-worker AppStderr C:\srv\app\logs\worker.log
nssm start redis-task-worker
```

`nssm stop redis-task-worker` and *Stop* in `services.msc` then go through
the graceful path. NSSM restarts the worker when it exits, so the loop keeps
running after a forced exit as well.

**Windows service with WinSW** - [WinSW](https://github.com/winsw/winsw)
sends Ctrl-C on stop and waits `stoptimeout` (15 seconds by default) before
killing the process:

```xml
<service>
  <id>redis-task-worker</id>
  <name>django-tasks-redis worker</name>
  <executable>C:\srv\app\venv\Scripts\python.exe</executable>
  <arguments>manage.py run_redis_tasks --continuous --shutdown-timeout=50</arguments>
  <workingdirectory>C:\srv\app</workingdirectory>
  <env name="DJANGO_SETTINGS_MODULE" value="myproject.settings"/>
  <stoptimeout>60 sec</stoptimeout>
  <log mode="roll"></log>
  <onfailure action="restart" delay="5 sec"/>
</service>
```

With either service manager, a system shutdown gives every service
`WaitToKillServiceTimeout` (20 seconds by default, under
`HKLM\SYSTEM\CurrentControlSet\Control`) before it is killed, whatever the
service's own timeout says. Keep `--shutdown-timeout` under that, or raise the
value, if tasks must survive a reboot without going through recovery.

### Using it in your own worker loop

The shutdown handling is available as a public API, for custom worker loops:

```python
from django_tasks_redis import GracefulShutdown, executor

with GracefulShutdown(timeout=50) as shutdown:
    while not shutdown.is_set():
        results = executor.process_tasks(max_tasks=10, stop_event=shutdown)
        if not results and shutdown.wait(5):  # interruptible sleep
            break
```

| API | Description |
| --------------------------------------------------------------------------------- | -------------------------------------------------------------- |
| `GracefulShutdown(signals=None, timeout=0, on_signal=None, force_on_repeat=True)` | Context manager that installs the signal handlers |
| `shutdown.is_set()` | True once a shutdown has been requested |
| `shutdown.wait(seconds)` | Sleep, returning early (True) when a shutdown is requested |
| `shutdown.set()` | Request a shutdown programmatically |
| `executor.process_tasks(..., stop_event=...)` | Stop starting new tasks once the event is set |
| `broker.receive(..., wait_seconds=...)` | Stops waiting early while the active `GracefulShutdown` is set |
| `is_shutdown_requested()` | True if the active worker was asked to shut down |

The same API, with the same names, is in django-database-task.

## Structured logging

The library logs to the `django_tasks_redis` logger and attaches its context as
record attributes rather than only baking it into the message, so a JSON
formatter emits fields you can filter on instead of one opaque string.

Every task record carries:

| Field | Value |
|-------|-------|
| `task_id` | The task's UUID, as a string |
| `task_path` | Dotted path of the task function |
| `queue_name` | Queue the task was enqueued on |
| `priority` | Priority it was enqueued with |
| `backend_alias` | Key in the `TASKS` setting |
| `worker_id` | `hostname-xxxxxxxx` of the worker that ran it |

Completed runs add `status` (`SUCCESSFUL` or `FAILED`) and `duration_ms`, the
wall time of the function call measured with `time.monotonic()` so it stays
accurate when a recovery sweep rewrites the stored timestamps. Failures also
add `error_class`. The worker's own start and finish records, and the record
for a read from the broker that failed, carry `worker_id`, `backend_alias`,
`queue_name`, and — on finish — `tasks_processed`, `tasks_failed` and
`exit_code`.

| Message | Level | When |
|---------|-------|------|
| `Worker started` | INFO | The command has resolved its backend |
| `Worker %s failed to receive a task` | ERROR | The read from the broker raised; no task was taken |
| `Task started` | INFO | Immediately before the task function is called |
| `Task completed successfully` | INFO | The task returned |
| `Task failed` | ERROR | The task raised |
| `Task could not be started` | ERROR | The task function could not be imported |
| `Worker %s failed to process a task` | ERROR | The worker could not run the task the message named |
| `Task abandoned` | ERROR | The queue gave up on the task (`mark_task_failed`) |
| `Worker finished` | INFO | The loop has ended, with the counts and exit code |

The standard library has no JSON formatter, so bring your own. This one has no
dependencies and merges whatever the library attached:

```python
# myproject/logging.py
import json
import logging

# Everything logging puts on a record by itself; the rest is ours.
_RESERVED = frozenset(vars(logging.LogRecord("", 0, "", 0, "", None, None))) | {
    "message",
    "asctime",
}


class JSONFormatter(logging.Formatter):
    def format(self, record):
        payload = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        payload.update({k: v for k, v in vars(record).items() if k not in _RESERVED})
        if record.exc_info:
            payload["traceback"] = self.formatException(record.exc_info)
        return json.dumps(payload, default=str)
```

```python
# settings.py
LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "json": {"()": "myproject.logging.JSONFormatter"},
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "json",
        },
    },
    "loggers": {
        "django_tasks_redis": {
            "handlers": ["console"],
            "level": "INFO",
            "propagate": False,
        },
    },
}
```

A completed task then reads:

```json
{"timestamp": "2026-09-12 10:00:01,123", "level": "INFO",
 "logger": "django_tasks_redis", "message": "Task completed successfully: id=... path=...",
 "task_id": "...", "task_path": "myapp.tasks.send_report",
 "queue_name": "reports", "priority": 0, "backend_alias": "default",
 "worker_id": "host-a1b2c3d4", "status": "SUCCESSFUL", "duration_ms": 42}
```

`duration_ms` is the same value the metrics integration from
[#2](https://github.com/tokibito/django-tasks-redis/pull/2) reads for its
duration histogram, so the backend is the single source of truth for how long
a task took.

## Running from a job scheduler

An on-premise scheduler — JP1, Hinemos, Rundeck, cron, a systemd timer — starts
`run_redis_tasks` on its own schedule, waits for it to exit, and decides
what happened from the exit code. That is a different shape from a long-running
worker, and two things make it work: exit codes the scheduler can act on, and
a lock so a slow run is not overlapped by the next one.

The scheduler is the trigger; the worker still reads the Redis stream the
backend writes to.

### Exit codes

By default the command exits 0 whether it ran a hundred tasks, none at all, or
one that failed — the same as before these options existed. Both options below
are opt-in, so adding them cannot break an existing `cron` line or Kubernetes
`Job`.

| Option | Meaning |
|--------|---------|
| `--empty-exit-code CODE` | Exit with `CODE` when no task was processed |
| `--failed-exit-code CODE` | Exit with `CODE` when at least one task failed or could not be run. Takes precedence over `--empty-exit-code` |

```bash
python manage.py run_redis_tasks --empty-exit-code=4 --failed-exit-code=1
```

With that line a scheduler sees:

| Exit code | What happened |
|-----------|---------------|
| `0` | At least one task ran and every one of them succeeded |
| `1` | At least one task failed, or the worker could not run it at all |
| `4` | There was nothing to do |
| `1` (without the options) | The command itself could not start — bad `--backend`, unreadable settings |

Pick the codes to suit the scheduler. JP1 compares the code against a warning
threshold per job, so an idle run is usually mapped to a warning code above the
normal end code and below the abnormal one; `--empty-exit-code=4` with a
warning threshold of 4 and an error threshold of 8 is a common arrangement.

Both codes must be between 0 and 255 — anything larger is truncated by the
operating system before the scheduler ever sees it, so the command rejects it
up front rather than reporting a code you did not choose.

A failure outranks an idle run. A task the worker could not run at all leaves
the processed count at zero while still being a failure, so both conditions can
hold at once, and `--failed-exit-code` wins.

What counts as a failure:

- a task that ran and ended `FAILED`
- a task the worker could not run at all (its code no longer imports, say,
  or `backend.run_task()` raised before the result was written)

What does not:

- a Redis error during `XREADGROUP` — a connection refused, a `NOGROUP` on
  startup, a network blip. Those are infrastructure faults rather than task
  outcomes; they are logged at `ERROR` but leave the exit code alone
- a broker message naming a task that no longer exists, or one another worker
  already holds. There was nothing for this worker to do

Tasks that failed are still recorded in Redis with their traceback, so a
nonzero exit is a prompt to look, not the report itself. `SIGTERM` during a run
is not an error: the worker finishes the task in hand and reports on what it
managed to process.

### One run at a time

Multiple workers are safe by design — tasks are claimed with the
`claim_task()` script, which checks the status and records the attempt
together, so two workers never run the same task. What a timer-driven setup
needs to avoid is different: a run that takes longer than the interval, with
the next launch piling on behind it until the host runs out of memory.

`flock(1)` handles that from outside, and needs nothing from this library:

```bash
flock -n --conflict-exit-code 3 /var/lock/redis-task-worker.lock \
    /srv/app/venv/bin/python manage.py run_redis_tasks \
        --empty-exit-code=4 --failed-exit-code=1
```

`-n` returns immediately instead of queueing behind the running process, and
`--conflict-exit-code 3` keeps "a run is already in progress" distinct from the
codes above — without it `flock` exits 1, which you cannot tell apart from a
failed task.

Use a lock file per queue if you run a job per queue, since the runs are
independent:

```bash
flock -n --conflict-exit-code 3 "/var/lock/redis-task-worker-$QUEUE.lock" \
    /srv/app/venv/bin/python manage.py run_redis_tasks --queue "$QUEUE"
```

The lock is about resource use on one host, not correctness. Workers on other
hosts hold their own lock files and still cannot collide over a task.

Windows has no `flock(1)`. Task Scheduler's *Do not start a new instance*
setting plays the same role there, see
[Windows Task Scheduler](#windows-task-scheduler); nothing is needed from the
library either way.

### systemd

Two shapes, depending on whether the worker stays up.

**Timer-driven** — the worker starts, drains the queue, and exits. This is the
equivalent of the cron / JP1 setup above, and the one to reach for when tasks
are infrequent.

`/etc/systemd/system/redis-task-worker.service`:

```ini
[Unit]
Description=Drain the django-tasks-redis queue
After=network-online.target redis-server.service

[Service]
Type=oneshot
User=app
WorkingDirectory=/srv/app
Environment=DJANGO_SETTINGS_MODULE=myproject.settings
ExecStart=/usr/bin/flock -n --conflict-exit-code 3 /var/lock/redis-task-worker.lock \
    /srv/app/venv/bin/python manage.py run_redis_tasks \
    --empty-exit-code=4 --failed-exit-code=1

# An idle run and an overlapping run are both expected, not failures.
SuccessExitStatus=3 4
```

`/etc/systemd/system/redis-task-worker.timer`:

```ini
[Unit]
Description=Drain the django-tasks-redis queue every minute

[Timer]
OnCalendar=*:0/1
# Do not fire a burst of catch-up runs after the host was asleep or down.
Persistent=false
AccuracySec=1s

[Install]
WantedBy=timers.target
```

```bash
systemctl enable --now redis-task-worker.timer
```

`SuccessExitStatus` is what stops systemd from logging an idle minute as a
failed unit. Leave the code for a failed task out of it, so `systemctl
--failed` and any alerting built on it still surface real problems.

**Long-running** — the worker stays up and polls. Prefer this when tasks arrive
continuously, or when you want a task picked up the moment it is enqueued.
Exit codes are close to meaningless here, since the process is not supposed to
exit; what matters is the shutdown timeout.

`/etc/systemd/system/redis-task-worker.service`:

```ini
[Unit]
Description=django-tasks-redis worker
After=network-online.target redis-server.service

[Service]
Type=simple
User=app
WorkingDirectory=/srv/app
Environment=DJANGO_SETTINGS_MODULE=myproject.settings
ExecStart=/srv/app/venv/bin/python manage.py run_redis_tasks \
    --continuous --shutdown-timeout=50
KillSignal=SIGTERM
# Longer than --shutdown-timeout, so the worker gets to finish its task.
TimeoutStopSec=60
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target
```

Run several by templating the unit (`redis-task-worker@.service` with
`--queue=%i`) rather than raising a concurrency setting — each process claims
its own tasks.

See [Graceful Shutdown](#graceful-shutdown) for what happens between `SIGTERM`
and `TimeoutStopSec`.

### Windows Task Scheduler

The timer-driven shape on Windows. The worker starts, drains the queue and
exits; Task Scheduler records the exit code as the task's *Last Run Result*
and in event 201 of its operational log, which is what a monitoring agent
reads. `-MultipleInstances IgnoreNew` is the *Do not start a new instance*
setting, and does what `flock -n` does above:

```powershell
$action = New-ScheduledTaskAction `
    -Execute "C:\srv\app\venv\Scripts\python.exe" `
    -Argument "manage.py run_redis_tasks --settings=myproject.settings --empty-exit-code=4 --failed-exit-code=1" `
    -WorkingDirectory "C:\srv\app"
$trigger = New-ScheduledTaskTrigger -Once -At (Get-Date) `
    -RepetitionInterval (New-TimeSpan -Minutes 1)
$settings = New-ScheduledTaskSettingsSet `
    -MultipleInstances IgnoreNew `
    -ExecutionTimeLimit (New-TimeSpan -Hours 1)
Register-ScheduledTask -TaskName "redis-task-worker" `
    -Action $action -Trigger $trigger -Settings $settings `
    -User "app" -Password $password
```

`--settings` stands in for `DJANGO_SETTINGS_MODULE`, since a scheduled task
cannot set environment variables of its own. The account needs a password so
the task runs whether or not anyone is logged on. Missed starts are skipped
unless *Run task as soon as possible after a scheduled start is missed* is
ticked, so there is no `Persistent=false` to set.

*Stop the task if it runs longer than* (`-ExecutionTimeLimit`) and *End task*
are hard kills, not a signal, so a run cut short that way leaves its task to
be reclaimed after `REDIS_CLAIM_TIMEOUT`. Set the limit well above the
longest run, or drop it with `-ExecutionTimeLimit (New-TimeSpan -Seconds 0)`.
The exit codes and what counts as a failure are the same as on Linux; see
[Exit codes](#exit-codes).

## 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.
The backend provides a list of handlers through `get_auth_handlers(endpoint)`.
Each handler takes a request and returns `None` to let it through, or a response
to refuse it. The handlers are tried in order and the request is accepted as
soon as one of them accepts it; when every handler rejects it, the first
rejection is returned. The endpoints are `csrf_exempt`, so the handlers are
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.

The simplest way to open the endpoints is `AUTH_HANDLERS` in the backend
`OPTIONS`:

```python
TASKS = {
    "default": {
        "BACKEND": "django_tasks_redis.RedisTaskBackend",
        "OPTIONS": {
            "AUTH_HANDLERS": [
                "django_tasks_redis.auth.SharedSecretAuth",
            ],
            "AUTH_HANDLER_OPTIONS": {"TOKEN_SETTING": "TASK_API_TOKEN"},
        },
    },
}
```

A project that needs more than a token overrides `get_auth_handlers()` instead
and returns its own list:

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

from django_tasks_redis.backends import RedisTaskBackend


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

        return [handler]
```

The bundled handlers cover the common cases:

- `SharedSecretAuth` — a bearer token (or any header value) compared with
  `hmac.compare_digest`. The token can be passed in `OPTIONS`, read from a
  Django setting with `TOKEN_SETTING`, or read from an environment variable
  with `TOKEN_ENV`. Reading it from a setting or an env var is preferred over
  writing it into `OPTIONS`.
- `HMACAuth` — a signature over `timestamp\nmethod\npath\nbody`, with replay
  protection (`MAX_AGE`, default 300 seconds, set to `0` to disable). Callers
  produce the signature with `django_tasks_redis.auth.build_signature()`.
- `StaffOnlyAuth` — accepts requests from a logged-in staff user; requires
  `django.contrib.auth.middleware.AuthenticationMiddleware`.

A handler entry in `AUTH_HANDLERS` can be a dotted path, a callable, an
instance, or a dict with `HANDLER`, `OPTIONS` and `ENDPOINTS`. The last limits
a handler to a subset of `run`, `run_one`, `status`, `execute` and `purge`,
so a cron job and the service that calls `execute/<id>/` can use different
credentials on the same backend:

```python
"AUTH_HANDLERS": [
    "django_tasks_redis.auth.SharedSecretAuth",
    {
        "HANDLER": "django_tasks_redis.auth.HMACAuth",
        "OPTIONS": {"SECRET_SETTING": "TASK_CRON_SECRET"},
        "ENDPOINTS": ["run", "purge"],
    },
],
```

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

### The stream broker

The functions above are thin wrappers over the backend's broker, which is
where reading, acknowledging and reclaiming messages live. It has the shape of
a pull broker in django-database-task, so a worker loop written against one
package reads the same against the other:

```python
from django.tasks import task_backends

backend = task_backends["default"]
broker = backend.broker  # a django_tasks_redis.brokers.RedisStreamsBroker

for message in broker.receive(
    queue_name="default", wait_seconds=5, worker_id=worker_id
):
    backend.run_task(message.task_id, worker_id=worker_id)
    broker.ack(message)
```

| Method | What it does on a Redis stream |
| -------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `receive(queue_name=None, max_messages=1, wait_seconds=0, worker_id=None)` | `XREADGROUP` as the consumer `worker_id`: the messages it already holds first, then new ones in priority order. Messages whose task is no longer `READY`, or whose task is gone, are acknowledged inside the call and not returned |
| `ack(message)` | `XACK` and `XDEL`. Until it is called the message stays pending for the consumer |
| `nack(message)` | Nothing. A pending entry is what a stream has instead of redelivery: the same consumer is served it again, or another worker takes it over once it has been idle for `REDIS_CLAIM_TIMEOUT` |
| `claim_stale_messages(worker_id, claim_timeout=None, max_deliveries=None)` | `XPENDING` and `XCLAIM`: take over what a dead consumer left, hand a task it left `RUNNING` back as `READY`, and give up on one started `REDIS_MAX_DELIVERIES` times. Consumers idle for the timeout that hold nothing are removed from the group |
| `remove_consumer(worker_id)` | `XGROUP DELCONSUMER` on every stream, for a worker on its way out. A consumer that still holds pending messages is kept for the sweep |

`worker_id` is the consumer name in the group, so it has to be the id the
worker keeps using: a message received as one consumer is only served again
to that consumer, or to whoever reclaims it.

There is no `notify()` step, unlike the brokers in django-database-task: the
backend writes to the stream when it enqueues, so the stream is the queue
rather than a notification about one. `broker_class` on the backend names
the class to build; a subclass of `RedisStreamsBroker` can change how any of
this is done.

## 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
