django-tasks-redis

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

• 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.)

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
    Backend->>Redis: ZADD to the READY status 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, moves the task in the status index,<br/>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<br/>(one transaction with the status index move)
    else Failure
        Worker->>Redis: HSET status=FAILED,<br/>errors, finished_at<br/>(one transaction with the status index move)
    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)

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

pip install django-tasks-redis

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

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
        },
    },
}

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.

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.

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

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)
  --task-path TASK_PATH   Only delete tasks with this task path
  --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)

Index every stored result under its status:

python manage.py rebuild_redis_status_index [options]

Options:
  --batch-size N          Tasks read per round trip (default: REDIS_SCAN_BATCH_SIZE)
  --backend BACKEND_NAME  Backend name (default: default)

The status index is what the task counts and the queue statistics are read from (see Monitoring). It is written with every task, so a new deployment never needs this command. A deployment that stored results before the index existed runs it once, after every process is on a version that writes the index, and reads its counts from the stored results until then. Workers can keep running while it does: a task that changes status under the rebuild is left to the write that changed it.

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

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.

$ 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).

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 SIGKILLed by the platform:

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 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 the handlers are installed for SIGINT, SIGTERM and SIGBREAK, but only two of them are ever delivered:

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

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:

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.

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

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:

[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):

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

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 sends Ctrl-C on stop and waits stoptimeout (15 seconds by default) before killing the process:

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

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

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

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

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:

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.

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

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

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

{"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 reads for its duration histogram, so the backend is the single source of truth for how long a task took.

Queue state and task duration are exposed without a dependency on a monitoring system, so a collector or a dashboard reads them through the library instead of the Redis key layout.

executor.get_queue_stats() returns the counts per status, the number of delayed tasks that have not come due, and the time the oldest and newest pending task started waiting: max(enqueued_at, run_after). How long the oldest pending task has been waiting is the signal a stuck queue gives: the counts can look healthy while nothing is serving the queue.

from django.utils import timezone

from django_tasks_redis import executor

stats = executor.get_queue_stats()
if stats["oldest_pending_waiting_since"]:
    age = timezone.now() - stats["oldest_pending_waiting_since"]
    print(f"oldest pending task has waited for {age.total_seconds():.0f}s")

With queue_name="emails" the numbers cover that queue alone. A pending task is one a worker would pick up now: a READY task whose run_after is unset or has passed. A delayed task whose time has not come is counted in delayed_count instead, so pending_count + delayed_count is the READY count of get_task_counts(), and a task that is not due yet does not read as queue age. get_pending_task_count() returns the same pending_count.

The same dict is served over HTTP by GET /tasks/status/ (see HTTP Endpoints), with the datetimes as ISO 8601 strings:

{"pending_count": 5, "running_count": 1, "successful_count": 120,
 "failed_count": 2, "delayed_count": 3,
 "oldest_pending_waiting_since": "2026-09-30T01:02:03.456Z",
 "newest_pending_waiting_since": "2026-09-30T01:04:05.678Z"}

The counts and the waiting times are read from a status index — one sorted set per status, for the backend and for each queue, holding the tasks in that status scored by the time they entered it — so a call costs the same however many results are stored, and a metrics scrape every few seconds does not keep Redis busy reading every result. Each status write moves the task in the index within the same script or transaction, so the index follows the hashes under concurrent workers, and an entry whose result has expired is dropped once it is older than the TTL of its status (REDIS_RESULT_TTL for READY and RUNNING, REDIS_COMPLETED_TASK_TTL for the rest). Shortening a TTL therefore takes results stored under the longer one out of the counts before they expire, and they stay listed in the admin until they do.

A deployment that stored results before the index existed keeps reading its counts from every stored result, as before, and logs a warning naming rebuild_redis_status_index (see Management Commands); run it once, after every process is on a version that writes the index, and the counts come from the index from then on. A new deployment needs nothing.

Task duration is read from Django's task_finished signal, which the backend sends with the finished TaskResult — for a run that failed as well as one that returned. finished_at minus last_attempted_at is the wall time of the last attempt:

# myproject/monitoring.py
from django.tasks.signals import task_finished


def record_task_duration(sender, task_result, **kwargs):
    if task_result.finished_at and task_result.last_attempted_at:
        duration = task_result.finished_at - task_result.last_attempted_at
        print(f"{task_result.id} took {duration.total_seconds():.3f}s")


task_finished.connect(record_task_duration)

Connect it where Django loads it, for example an app config's ready(). The wall time is measured from the claim, so it is not the same number as the duration_ms on the log records (see the Structured logging section), which is the time the task function itself spent running, measured with time.monotonic(). The signal is not sent when the queue gives up on a task without running it (mark_task_failed), nor when run_task cannot resolve the task object after the claim and fails the task: a duration monitor does not see those tasks, they show up in the failed count and in the Task abandoned or Task could not be started log records instead.

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.

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.

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

With that line a scheduler sees:

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.

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:

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:

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; nothing is needed from the library either way.

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:

[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:

[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

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:

[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 for what happens between SIGTERM and TimeoutStopSec.

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:

$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.

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

• View task list with status, priority, queue
• Filter the task list by status, queue, task path, or priority
• 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.

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.

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/ - Get the queue statistics of get_queue_stats() (?queue_name= for one queue)
• 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:

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:

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:

"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.

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

# Get queue statistics
stats = executor.get_queue_stats()

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

# Purge one task path only
deleted = executor.purge_completed_tasks(days=7, task_path="myapp.tasks.cleanup")

# Index the results stored before the status index existed (once, after upgrading)
indexed = executor.rebuild_status_index()

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:

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)

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.

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

MIT License