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

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}

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.

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.
• No priorities. SQS has none. Use separate queues and give the important ones more workers.
• 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