Metadata-Version: 2.3
Name: taskiq-sqs
Version: 0.1.0
Summary: SQS Broker for TaskIQ
Keywords: taskiq,broker,aws,sqs
License: MIT License
         
         Copyright (c) 2026 Mike Shultz and Taskiq team
         
         Permission is hereby granted, free of charge, to any person obtaining a copy
         of this software and associated documentation files (the "Software"), to deal
         in the Software without restriction, including without limitation the rights
         to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
         copies of the Software, and to permit persons to whom the Software is
         furnished to do so, subject to the following conditions:
         
         The above copyright notice and this permission notice shall be included in all
         copies or substantial portions of the Software.
         
         THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
         IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
         FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
         AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
         LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
         OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
         SOFTWARE.
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Information Technology
Classifier: Natural Language :: English
Classifier: Framework :: AsyncIO
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: Software Development
Classifier: Topic :: System :: Networking
Classifier: Typing :: Typed
Classifier: Operating System :: OS Independent
Requires-Dist: taskiq>=0.12.6
Requires-Dist: capo-s3>=0.15.0
Requires-Dist: capo-sqs>=0.6.0
Requires-Python: >=3.11
Project-URL: Source, https://github.com/taskiq-python/taskiq-sqs
Project-URL: Bug Tracker, https://github.com/taskiq-python/taskiq-sqs/issues
Project-URL: Repository, https://github.com/taskiq-python/taskiq-sqs/
Project-URL: Documentation, https://taskiq-python.github.io/taskiq-sqs/
Description-Content-Type: text/markdown

# taskiq-sqs

[![PyPI - Python Version](https://img.shields.io/pypi/pyversions/taskiq-sqs?style=for-the-badge&logo=python)](https://pypi.org/project/taskiq-sqs/)
[![PyPI](https://img.shields.io/pypi/v/taskiq-sqs?style=for-the-badge&logo=pypi)](https://pypi.org/project/taskiq-sqs/)
[![Checks](https://img.shields.io/github/check-runs/taskiq-python/taskiq-sqs/main?nameFilter=test%20(ubuntu-latest,%203.12)&style=for-the-badge)](https://github.com/taskiq-python/taskiq-sqs)

This library provides SQS broker and S3 result backend for TaskIQ.

## Installation

```bash
pip install taskiq-sqs
```

## Basic usage

Here is an example of how to use the SQS broker with the S3 backend:

```python
import asyncio
from taskiq_sqs import S3ResultBackend, SQSBroker
from taskiq_sqs.types import S3Bucket, SQSQueue

broker = SQSBroker(
    queues=SQSQueue(name="my-queue"),  # by default the broker creates the queue for you if it doesn't exist
    endpoint_url="http://localhost:4566",
    aws_region_name="us-east-1",
).with_result_backend(
    S3ResultBackend(
        bucket=S3Bucket(name="response-bucket")  # by default backend will create bucket for you if it does not exist
    )
)

@broker.task()
async def i_love_aws() -> None:
    await asyncio.sleep(1)
    print("Hello there!")

async def main() -> None:
    await broker.startup()
    task = await i_love_aws.kiq()
    print(await task.wait_result())
    await broker.shutdown()

if __name__ == "__main__":
    asyncio.run(main())
```

How to run:
- run worker first with `taskiq worker examples.example_broker:broker`
- after that run broker to create a task and wait for result: `python examples/example_broker.py`

## Multiple queues

`SQSBroker` accepts a single queue or a list of them. The first queue is the default one, used whenever a task doesn't say otherwise. To send a task to a specific queue, set the `queue_name` label with that queue's name:

```python
from taskiq_sqs import SQSBroker
from taskiq_sqs.types import SQSQueue

broker = SQSBroker(
    queues=[
        SQSQueue(name="default-queue"),
        SQSQueue(name="high-priority-queue", wait_time_seconds=5),
    ],
)

@broker.task(queue_name="high-priority-queue")
async def urgent_task() -> None:
    ...
```

A worker started against this broker consumes from every configured queue at once. Passing a queue name through the `queue_name` label that isn't configured on the broker raises `UnknownQueueError`.

## Declaring queues

By default the broker creates a queue on startup if it doesn't exist yet, the same way `S3Bucket` does for buckets. Set `is_declare=False` to require the queue to already exist instead (raises `QueueNotFoundError` if it doesn't). `options` are queue attributes (e.g. `VisibilityTimeout`, `MessageRetentionPeriod`) passed to `CreateQueue`, in AWS's own PascalCase naming, when the queue is declared — they have no effect on a queue that already exists:

```python
from taskiq_sqs import SQSBroker
from taskiq_sqs.types import SQSQueue

broker = SQSBroker(
    queues=SQSQueue(name="my-queue", options={"VisibilityTimeout": "60", "MessageRetentionPeriod": "86400"}),
)
```

FIFO queues get their `FifoQueue` attribute set automatically when declared — no need to include it in `options`.

`S3Bucket` has the same `options` field, for parameters `CreateBucket` accepts beyond `name` (e.g. `acl`), passed through whenever `S3ResultBackend`/`S3OffloadMiddleware` create the bucket.

## Delayed tasks

Set the `delay` label to delay delivery of a task by that many seconds (0-900, SQS's own limit)

```python
from taskiq_sqs import SQSBroker
from taskiq_sqs.types import SQSQueue

broker = SQSBroker(queues=SQSQueue(name="my-queue"))

@broker.task(delay=30)  # "delay" is taskiq_sqs.constants.SQS_DELAY_SECONDS_LABEL
async def send_reminder() -> None:
    ...
```

A value outside the 0-900 range (or not an integer) raises `InvalidDelaySecondsError` when the task is kicked.

## FIFO queues

A queue whose name ends in `.fifo` is treated as a FIFO queue automatically, matching SQS's own naming rule (`SQSQueue`'s `is_fifo` field only needs to be set to override that default, and the queue's name must still end in `.fifo` for the broker to accept it as FIFO).

```python
from taskiq_sqs import SQSBroker
from taskiq_sqs.types import SQSQueue

broker = SQSBroker(queues=SQSQueue(name="my-queue.fifo"))

@broker.task(group_id="orders")  # defaults to the task name if not set
async def process_order() -> None:
    ...
```

- `group_id` picks the message's `MessageGroupId` (required by SQS for every FIFO message); it defaults to the task's name.
- `deduplication_id` sets `MessageDeduplicationId`; if not set, the queue must have content-based deduplication enabled, or SQS rejects the message.
- The `delay` label (see [Delayed tasks](#delayed-tasks)) is not supported on FIFO queues — SQS only allows delay to be configured on the queue itself, not per message — and raises `FifoDelayNotSupportedError` if used.

## Message expiration

Set the `expiry` label to a unix timestamp; if a worker receives the message after that time, it's deleted without being executed:

```python
import time
from taskiq_sqs import SQSBroker
from taskiq_sqs.types import SQSQueue

broker = SQSBroker(queues=SQSQueue(name="my-queue"))

@broker.task()
async def process_event() -> None:
    ...

await process_event.kicker().with_labels(expiry=time.time() + 300).kiq()  # discarded if received after 5 minutes
```

Expiration is checked by the worker on receipt, not by SQS itself — a message can still sit in the queue past its expiry (e.g. while workers are busy or scaled to zero), it just won't run once picked up. `expiry` must be a non-negative number; anything else raises `InvalidExpiryError` when the task is kicked.

## Message batching

Set `is_batching_enabled` on a queue to buffer kicked messages in memory and flush them together via `SendMessageBatch` (up to `batch_size` messages, or after `batch_timeout` seconds, whichever comes first) instead of sending each one immediately:

```python
from taskiq_sqs import SQSBroker
from taskiq_sqs.types import SQSQueue

broker = SQSBroker(
    queues=SQSQueue(name="my-queue", is_batching_enabled=True, batch_size=10, batch_timeout=1.0),
)
```

## Offloading large messages to S3

SQS messages are limited to 256 KiB. `S3OffloadMiddleware` transparently uploads task payloads that exceed a configurable threshold to S3 before sending them to the queue, and replaces the message with a reference to the uploaded object. The worker downloads the original payload back from S3 before executing the task, and (by default) removes it from S3 afterwards.

```python
import asyncio
from taskiq_sqs import S3OffloadMiddleware, SQSBroker
from taskiq_sqs.types import S3Bucket, SQSQueue

broker = SQSBroker(queues=SQSQueue(name="my-queue"))
broker.add_middlewares(
    S3OffloadMiddleware(
        bucket=S3Bucket(name="offload-bucket"),  # created automatically if it doesn't exist
        max_message_size=200_000,  # payloads larger than this many bytes are offloaded to S3
    ),
)

@broker.task
async def process_document(content: str) -> int:
    return len(content)


async def main() -> None:
    await broker.startup()
    await process_document.kiq("x" * 1_000_000)  # too large for SQS, transparently offloaded to S3
    await broker.shutdown()

if __name__ == "__main__":
    asyncio.run(main())
```
