Metadata-Version: 2.5
Name: django-tasks-sqs
Version: 0.1.0
Summary: Amazon SQS backend and worker for Django's built-in tasks framework (django.tasks)
Project-URL: Homepage, https://github.com/mgarcia0094/django-tasks-sqs
Project-URL: Repository, https://github.com/mgarcia0094/django-tasks-sqs
Project-URL: Issues, https://github.com/mgarcia0094/django-tasks-sqs/issues
Project-URL: Changelog, https://github.com/mgarcia0094/django-tasks-sqs/blob/main/CHANGELOG.md
Author-email: Miguel Ángel García Gandía <mgarcia0094@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: aws,background-jobs,django,queue,sqs,tasks,worker
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 :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Python: >=3.12
Requires-Dist: boto3>=1.35
Requires-Dist: django>=6.0
Description-Content-Type: text/markdown

# django-tasks-sqs

[![CI](https://github.com/mgarcia0094/django-tasks-sqs/actions/workflows/ci.yml/badge.svg)](https://github.com/mgarcia0094/django-tasks-sqs/actions/workflows/ci.yml)
[![PyPI](https://img.shields.io/pypi/v/django-tasks-sqs)](https://pypi.org/project/django-tasks-sqs/)
[![Python](https://img.shields.io/pypi/pyversions/django-tasks-sqs)](https://pypi.org/project/django-tasks-sqs/)
[![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE)

An **Amazon SQS backend and worker** for Django's built-in
[tasks framework](https://docs.djangoproject.com/en/stable/topics/tasks/) (`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.

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

```bash
pip install django-tasks-sqs
```

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

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

```json
{"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

| Module | What lives there |
|---|---|
| [`backend.py`](src/django_tasks_sqs/backend.py) | `SQSBackend`: settings, boto3 client, queue URL resolution, `enqueue`, system checks |
| [`message.py`](src/django_tasks_sqs/message.py) | `TaskMessage`: the JSON envelope and its validation |
| [`worker.py`](src/django_tasks_sqs/worker.py) | `Worker`: polling threads, execution, retries, heartbeat, deferral |
| [`management/commands/sqs_worker.py`](src/django_tasks_sqs/management/commands/sqs_worker.py) | CLI flags and signal handling |

## Worker options

| Flag | Default | |
|---|---|---|
| `--backend` | `default` | Alias in `settings.TASKS` |
| `--queue` | all `QUEUES` | Repeat to consume several |
| `--concurrency` | 1 | Polling threads per queue, each running one task at a time |
| `--wait-time` | 20 | Long-poll seconds (max 20) |
| `--max-messages` | 1 | Messages per receive (1–10). Keep it low for slow tasks |
| `--visibility-timeout` | queue's | Override for received messages |
| `--retry-backoff` | off | Base seconds for exponential backoff: `base * 2**(attempt-1)` |
| `--no-heartbeat` | | Don't extend visibility while tasks run |

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](https://www.localstack.cloud/) or
[moto](https://docs.getmoto.org/) in server mode:

```python
"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](CONTRIBUTING.md). In short:

```bash
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](CODE_OF_CONDUCT.md). To report a security issue,
see [SECURITY.md](SECURITY.md).

## License

[MIT](LICENSE)
