django-tasks-sqs

An Amazon SQS backend and worker for Django's built-in tasks framework (django.tasks, Django 6.0+).

Django defines how you declare and enqueue background tasks, but ships no production backend and no worker. This package provides both, on top of SQS. Your code only uses the standard django.tasks API, so you can switch backends later without touching it.

from django.tasks import task

@task
def send_welcome_email(user_id: int) -> None:
    ...

send_welcome_email.enqueue(user_id=42)   # returns immediately; a worker runs it

Install

pip install django-tasks-sqs

# settings.py
INSTALLED_APPS = [..., "django_tasks_sqs"]

TASKS = {
    "default": {
        "BACKEND": "django_tasks_sqs.SQSBackend",
        "QUEUES": ["default", "emails"],
        "OPTIONS": {
            "region_name": "eu-west-1",
            # SQS queue = prefix + django queue name ("myapp-default", "myapp-emails")...
            "queue_name_prefix": "myapp-",
            # ...or map queue names to URLs explicitly:
            # "queue_urls": {"default": "https://sqs.eu-west-1.amazonaws.com/123456789012/app"},
        },
    }
}

Then run one or more workers:

python manage.py sqs_worker                       # all queues of the "default" backend
python manage.py sqs_worker --queue emails --concurrency 4

The queues must already exist. Create them with your usual infrastructure tooling (Terraform, CDK, CloudFormation…).

Features

• Standard django.tasks API: @task, .enqueue(), .aenqueue(), .using(), takes_context, async tasks, and the task_enqueued / task_started / task_finished signals.
• Deferred tasks: task.using(run_after=...). Delays up to 15 minutes use SQS DelaySeconds. Longer delays are re-queued by the worker until they are due.
• Retries and dead-letter queues: a failed task is not deleted, so SQS delivers it again. Add a redrive policy to the queue to move it to a DLQ after maxReceiveCount attempts. --retry-backoff N adds exponential backoff between attempts.
• Long-running tasks: a heartbeat extends the message's visibility timeout while the task runs, so another worker doesn't pick it up halfway through.
• Graceful shutdown: on SIGTERM/SIGINT the worker finishes its current tasks and exits. Plays well with ECS, Kubernetes and systemd.
• Priorities: task.using(priority=...) routes to one SQS queue per priority level, and workers poll the higher levels more often (see below).
• Batch enqueue: backend.enqueue_many(...) sends many tasks with SendMessageBatch (see below).
• Health checks and metrics: --health-port serves /healthz and Prometheus /metrics; the message_processed signal feeds any other metrics system.
• FIFO queues (queue names ending in .fifo).
• Typed, with 100% test coverage. Tested against Django 6.0 and 6.1 on Python 3.12–3.14.

Enqueueing many tasks at once

django.tasks has no bulk API, so the backend adds one. enqueue_many takes (task, args, kwargs) tuples and sends them with as few SendMessageBatch requests as possible (up to 10 messages and 256 KiB per request, one queue per request):

backend = send_welcome_email.get_backend()
results = backend.enqueue_many(
    (send_welcome_email, [], {"user_id": user.pk}) for user in new_users
)

Every call is validated before anything is sent, and the TaskResults come back in the same order. task_enqueued is sent for each task. If SQS rejects some messages, the rest are still sent and django_tasks_sqs.EnqueueBatchError is raised: its enqueued lists the tasks that went through, and failed pairs each rejected task with SQS's error. aenqueue_many is the async version.

How it works

 web process                        SQS queue                     sqs_worker
 ───────────                        ─────────                     ──────────
 task.enqueue(args)  ──JSON msg──▶  [ ... ]  ──long poll──▶  import task by path
                                                                   │
                                            delete ◀── success ────┤
                        visibility timeout expires ◀── failure ────┘ (retry / DLQ)

1. Enqueueing

task.enqueue(*args, **kwargs) runs in your web process. SQSBackend:

1. validates the task, as django.tasks requires: module-level function, JSON-serialisable arguments, a queue listed in QUEUES…;
2. resolves the SQS queue URL, either from queue_urls or by calling GetQueueUrl with queue_name_prefix + queue_name (the result is cached);
3. sends one message and returns a TaskResult with status READY.

The message body is a small, versioned JSON envelope. Only the task's import path travels, never code or pickles:

{"v": 1, "id": "…", "task": "myapp.tasks.send_welcome_email",
 "args": [], "kwargs": {"user_id": 42}, "queue_name": "default",
 "backend": "default", "enqueued_at": "2026-09-25T10:00:00+00:00", "run_after": null, "priority": 0}

The task path is also sent as a message attribute (task), which is handy for filtering and debugging in the AWS console.

2. Deferring (run_after)

SQS can delay a message for at most 15 minutes (DelaySeconds). For longer delays the backend sends the message with the maximum delay. When a worker receives it before run_after, it sends a new copy delayed again and deletes the original. This repeats until the task is due, so a task can be deferred for any length of time.

3. Consuming

manage.py sqs_worker starts --concurrency threads per queue. Each thread loops:

1. Long-polls SQS (ReceiveMessage with WaitTimeSeconds=20), which is cheap when the queue is idle.
2. Imports the task by path. If the message is malformed or the task can't be imported, it is logged and left alone, so the redrive policy eventually moves it to the DLQ.
3. Runs it exactly like Django's ImmediateBackend does: it builds a TaskResult, sends task_started, calls the function (sync or async, with TaskContext if takes_context=True), sends task_finished, and closes stale DB connections before and after.
4. Acknowledges or retries. On success the message is deleted. On failure it is kept: SQS delivers it again when the visibility timeout expires, or after --retry-backoff seconds (doubling each attempt) if you set it.

While a task runs, a heartbeat thread calls ChangeMessageVisibility every half timeout, so a slow task is never handed to a second worker.

4. Shutting down

SIGTERM/SIGINT sets a stop flag. Threads finish the task they are running, stop polling and exit. A message that was received but not finished simply becomes visible again, and another worker picks it up.

Code map

Worker options

You can also run a worker from code with django_tasks_sqs.Worker.

Health checks and metrics

python manage.py sqs_worker --health-port 8000

• GET /healthz returns 200 ok or 503 unhealthy. The worker is healthy when every polling thread either received from SQS successfully in the last --health-max-age seconds or is busy running a task. So a long task never fails the check, but a worker stuck retrying (expired credentials, no network) does. Point a Kubernetes liveness probe or an ECS health check (curl -f) at it.

• GET /metrics uses the Prometheus text format:

  Metric	Type
  django_tasks_sqs_messages_total{queue,outcome}	counter	succeeded, failed, deferred, invalid
  django_tasks_sqs_poll_errors_total{queue}	counter	Failed ReceiveMessage calls
  django_tasks_sqs_busy_threads	gauge	Threads running a task
  django_tasks_sqs_last_poll_age_seconds	gauge	Seconds since the stalest idle thread polled
  django_tasks_sqs_healthy	gauge	1 or 0, as /healthz

  queue is the SQS queue name without prefix, including the priority suffix if any (emails-high). The same goes for queue_name in the signal below.

For CloudWatch, StatsD or anything else, connect to the message_processed signal. It is sent after every message with worker, queue_name, outcome and duration (seconds):

from django.dispatch import receiver
from django_tasks_sqs.signals import message_processed

@receiver(message_processed)
def record(sender, queue_name, outcome, duration, **kwargs):
    statsd.timing(f"tasks.{queue_name}.{outcome}", duration * 1000)

When running a Worker from code, the same data is in worker.stats, and django_tasks_sqs.health.HealthServer(worker, port=...) serves the endpoints.

Priorities

SQS has no priorities, so the backend emulates them with one SQS queue per priority level. Configure the levels, highest first, as (min_priority, suffix):

"OPTIONS": {
    "queue_name_prefix": "myapp-",
    "priority_levels": [
        (50, "-high"),   # priority >= 50   -> myapp-emails-high
        (0, ""),         # 0 <= priority < 50 -> myapp-emails
        (-100, "-low"),  # priority < 0     -> myapp-emails-low
    ],
},

send_email.using(priority=80).enqueue(...)   # goes to myapp-emails-high

• Every queue in QUEUES gets every level, and all those SQS queues must exist. For FIFO queues the suffix goes before .fifo (orders-high.fifo). With queue_urls, use the suffixed names as keys ("emails-high": "https://…").
• A priority lower than every min_priority goes to the last level.
• Without priority_levels, the backend keeps supports_priority = False and rejects tasks with a non-default priority, as before.

Polling. Each worker thread tries the levels of its queue in a random order weighted by each level's weight. It checks all but the last without waiting and long-polls the last one. Weights default to powers of two (4, 2, 1 for three levels), so the top level is checked first 4 times out of 7. Set them explicitly with a third element: (50, "-high", 10). Low levels are never starved, but a high-priority task can wait up to --wait-time seconds while a thread long-polls a lower level. Lower --wait-time if that matters more to you than the number of receive calls.

Deferred tasks stay on their level when the worker re-sends them.

Things to know

• Delivery is at least once. That is how SQS works: a task can run more than once, for example if a worker dies after running it but before deleting the message. Make tasks idempotent.
• No result storage (yet). supports_get_result = False, so task.get_result(id) raises NotImplementedError. Store results yourself if you need them.
• FIFO queues don't support run_after, because SQS has no per-message delay on FIFO queues.
• Messages are limited to 256 KB. Pass IDs, not big payloads.
• TaskContext.attempt comes from SQS's ApproximateReceiveCount.
• IAM permissions: the web process needs sqs:SendMessage and sqs:GetQueueUrl. Workers also need sqs:ReceiveMessage, sqs:DeleteMessage, sqs:ChangeMessageVisibility and sqs:GetQueueAttributes.

Local development

Point endpoint_url at LocalStack or moto in server mode:

"OPTIONS": {"endpoint_url": "http://localhost:4566", "region_name": "us-east-1", ...}

For unit tests, use Django's ImmediateBackend instead, which runs tasks inline.

Roadmap

Ideas where help is very welcome. Open an issue to discuss before starting something big:

• Optional result storage (e.g. in the Django database), so get_result() works
• Batch sends (SendMessageBatch) for enqueueing many tasks at once
• Priorities emulated with several queues and weighted polling
• Health check and metrics hooks for the worker (Prometheus / CloudWatch)
• Payloads over 256 KB stored in S3 (extended client pattern)
• Integration tests against LocalStack in CI

Contributing

Contributions of any size are welcome: bug reports, docs, tests, features. Start with CONTRIBUTING.md. In short:

git clone https://github.com/mgarcia0094/django-tasks-sqs && cd django-tasks-sqs
uv sync              # install with dev dependencies
uv run pytest        # tests (SQS is mocked with moto, no AWS account needed)
uv run ruff check .  # lint
uv run mypy          # strict type checking

Please follow the Code of Conduct. To report a security issue, see SECURITY.md.

License

MIT