TaskIQ SQS/S3 aiobotocore

This library provides you with a fully asynchronous SQS broker and S3 backend for TaskIQ using aiobotocore. Inspired by the taskiq-sqs broker.

Key Features

• Async SQS Broker: Fully asynchronous SQS message broker with support for standard and FIFO queues (see General Usage)
• Multi-Queue Routing: Register multiple queues and route tasks per message/task label (see Multi-Queue Usage)
• S3 Result Backend: Store task results in S3, ideal for large result payloads (see General Usage)
• Extended Messages: Automatic S3 storage for messages exceeding SQS limits (see Extended Messages with S3)
• Message Batching: Improved performance + cost reduction through batching multiple messages in single SQS operations (see Message Batching)
• Delayed Tasks: Schedule tasks with configurable delays (0-900 seconds) (see Delayed Tasks)
• FIFO Queue Support: Message ordering and deduplication with custom MessageGroupId control per task/message (see FIFO Queues and Custom Message Groups)
• Fair Queues: Distribute tasks evenly across message groups for balanced processing (see Configuration)
• Message Metadata: Optionally surface the SQS receipt handle, message id, and approximate receive count to task code, useful for debugging duplicate/redelivered messages (see Message Metadata)
• Heartbeat: Optionally keep long-running tasks' messages invisible by periodically extending their visibility timeout in the background, preventing premature redelivery/duplicate processing (see Heartbeat)

Installation

pip install taskiq-aio-sqs

Requirements:

• Pydantic v2 (pydantic>=2.0,<3.0)

General Usage:

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

# broker.py
import asyncio
from taskiq_aio_sqs import SQSBroker, S3Backend

s3_result_backend = S3Backend(
    bucket_name="response-bucket",  # bucket must exist
)

broker = SQSBroker(
    sqs_queue_name="my-queue",
).with_result_backend(s3_result_backend)


@broker.task()
async def i_love_aws() -> None:
    """I hope my cloud bill doesn't get too high!"""
    await asyncio.sleep(5.5)
    print("Hello there!")


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


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

Multi-Queue Usage:

Use SQSQueue definitions with with_queues / with_default_queue before startup, then route tasks using the queue label.

from taskiq_aio_sqs import SQSBroker, SQSQueue

broker = SQSBroker(
    sqs_queue_name="taskiq-default",
)

critical_queue = SQSQueue(name="taskiq-critical")
fifo_queue = SQSQueue(name="taskiq-events.fifo", is_fifo=True)

broker.with_queues(critical_queue, fifo_queue)


@broker.task(queue="taskiq-critical")
async def critical_task() -> None:
    pass


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


async def publish_examples() -> None:
    await broker.startup()

    # Uses default queue (taskiq-default)
    await normal_task.kiq()

    # Uses queue from task decorator
    await critical_task.kiq()

    # Per-call queue override
    await normal_task.kicker().with_labels(queue="taskiq-critical").kiq()

    # FIFO queue with explicit group
    await normal_task.kicker().with_labels(
        queue="taskiq-events.fifo",
        group_id="orders",
    ).kiq()

    await broker.shutdown()

Notes:

• Queue names must be registered with with_queues (or be the default queue) before startup().
• listen() consumes from all configured queues.

Delayed Tasks:

Delayed tasks can be created in 3 ways:

• by using the delay parameter in the task decorator
• by using the kicker with the delay label
• by setting the delay_seconds parameter in the broker, which will apply to all tasks processed by the broker.

Here's an example of how to use delayed tasks:

broker = SQSBroker(
    delay_seconds=3,
    sqs_queue_name="my-queue",
)

@broker.task()
async def general_task() -> int:
    return 1

@broker.task(delay=7)
async def delayed_task() -> int:
    return 1

async def main():
    await broker.startup()
    # This message will be received by workers after 3 seconds
    # delay using the delay_seconds parameter in the broker init.
    await general_task.kiq()

    # This message will be received by workers after 7 seconds delay.
    await delayed_task.kiq()

    # This message is going to be received after the delay in 4 seconds.
    # Since we override the `delay` label using kicker.
    await delayed_task.kicker().with_labels(delay=4).kiq()

Message Batching:

The SQS broker supports message batching to improve throughput and reduce AWS API calls. When batching is enabled, the broker collects multiple messages and sends them in a single SQS batch operation (up to 10 messages per batch).

Key Benefits:

• Improved Performance: Reduced number of API calls to SQS
• Cost Optimization: Fewer SQS requests means lower AWS costs
• Better Throughput: Can send up to 10 messages in a single operation

Here's an example of how to enable and configure message batching:

broker = SQSBroker(
    sqs_queue_name="my-queue",
    enable_batching=True,
    batch_size=5,           # Send batches when 5 messages are collected
    batch_timeout=2.0,      # Or send after 2 seconds, whichever comes first
    skip_batch_tasks=["urgent_task"],  # These tasks bypass batching
)

@broker.task()
async def normal_task(data: str) -> str:
    return f"Processed: {data}"

@broker.task()
async def urgent_task(alert: str) -> str:
    # This task bypasses batching due to skip_batch_tasks configuration
    return f"URGENT: {alert}"

async def main():
    await broker.startup()

    # These messages will be batched together
    await normal_task.kiq("message 1")
    await normal_task.kiq("message 2")
    await normal_task.kiq("message 3")
    # Batch will be sent when batch_size (5) is reached or batch_timeout (2s) expires

    # This message bypasses batching and is sent immediately
    await urgent_task.kiq("System alert!")

    # You can also bypass batching for individual tasks using the skip_batching label
    await normal_task.kicker().with_labels(skip_batching=True).kiq("priority message")

Batching Configuration:

• enable_batching: Enable/disable message batching (default: False)
• batch_size: Maximum messages per batch, 1-10 (default: 10)
• batch_timeout: Maximum wait time in seconds before sending partial batch, ≥0.1 (default: 1.0)
• skip_batch_tasks: List of task names that should always bypass batching (default: [])

Task-Level Control:

• Use the skip_batching=True label to bypass batching for specific task calls
• Tasks listed in skip_batch_tasks always bypass batching

Important Notes:

• Batching works with both standard and FIFO queues
• Messages in the same batch will have the same MessageGroupId when using FIFO queues
• Messages with per-message delay labels are sent immediately (not batched)
• Messages using S3 extended payload flow are sent immediately (not batched)
• The broker ensures all messages are sent when shutting down, even partial batches

Extended Messages with S3:

You can also use S3 to store messages that are too large for SQS. To do this, you need to set the s3_extended_bucket_name parameter in the broker configuration.

Here's an example of this behaviour:

pub_broker = SQSBroker(
    sqs_queue_name="my-queue",
    s3_extended_bucket_name="response-bucket",
)

sub_broker = SQSBroker(
    sqs_queue_name="my-queue",
)

LARGE_MESSAGE = b"x" * (1024 * 1024 + 1)  # 1 MB is the new limit for SQS

@pub_broker.task()
async def large_task() -> bytes:
    return LARGE_MESSAGE


async def main():
    await pub_broker.startup()
    await sub_broker.startup()
    # This message will store data in S3 and send a reference to SQS
    # This reference will include the S3 bucket and key.
    await large_task.kiq()

    async for msg in sub_broker.listen():
        message = msg
        break  # Stop after receiving one message

    # The message will be automatically retrieved from S3
    # and the full data will be available in the message.
    assert message.data == LARGE_MESSAGE

FIFO Queues and Custom Message Groups:

When using FIFO queues (queue names ending with .fifo), you can control message ordering by setting custom MessageGroupId values. Messages with the same MessageGroupId are processed in strict FIFO order, while messages with different MessageGroupId values can be processed in parallel.

Here's an example of how to use custom MessageGroupId:

broker = SQSBroker(
    sqs_queue_name="my-queue.fifo",  # FIFO queue
    use_task_id_for_deduplication=True,  # Recommended for FIFO
)

@broker.task()
async def process_user_action(user_id: int, action: str) -> str:
    # Process user action in order per user
    return f"Processed {action} for user {user_id}"

async def main():
    await broker.startup()

    # These tasks will be processed in order for each user,
    # but different users can be processed in parallel
    await process_user_action.kicker().with_labels(
        group_id="user_123"
    ).kiq(user_id=123, action="login")

    await process_user_action.kicker().with_labels(
        group_id="user_123"
    ).kiq(user_id=123, action="update_profile")

    await process_user_action.kicker().with_labels(
        group_id="user_456"
    ).kiq(user_id=456, action="purchase")

MessageGroupId Rules:

• Must be 1-128 characters long
• Can contain alphanumeric characters and punctuation: !"#$%&'()*+,-./:;<=>?@[\]^_{|}~`
• If no group_id label is provided, the task name will be used as MessageGroupId
• Tasks with the same MessageGroupId are processed sequentially
• Tasks with different MessageGroupId values can be processed in parallel

Note: Delay functionality is not supported with FIFO queues due to AWS SQS limitations.

Message Metadata:

SQS gives you the receipt handle, message id, and approximate receive count directly on every receive_message response. The catch is that taskiq's core Receiver only ever hands your task the raw message bytes, it has no concept of broker-specific delivery metadata, so out of the box there's no way for a task to know which physical SQS delivery it came from. That makes it impossible to distinguish "SQS redelivered this message" (a growing ApproximateReceiveCount) from "the publisher produced a duplicate message", which matters a lot when debugging duplicate task execution.

Set expose_message_metadata=True on the broker to opt into surfacing this metadata as labels on TaskiqMessage, readable from Context in your task:

from taskiq import Context, TaskiqDepends
from taskiq_aio_sqs import (
    SQSBroker,
    SQS_RECEIPT_HANDLE_LABEL,
    SQS_APPROXIMATE_RECEIVE_COUNT_LABEL,
)

broker = SQSBroker(
    sqs_queue_name="my-queue",
    expose_message_metadata=True,
)


@broker.task()
async def my_task(context: Context = TaskiqDepends()) -> None:
    receipt_handle = context.message.labels.get(SQS_RECEIPT_HANDLE_LABEL)
    receive_count = context.message.labels.get(SQS_APPROXIMATE_RECEIVE_COUNT_LABEL)
    # e.g. log both alongside your task_id to tell redeliveries apart
    # from publisher-side duplicates.

Available labels: SQS_RECEIPT_HANDLE_LABEL, SQS_QUEUE_URL_LABEL, SQS_MESSAGE_ID_LABEL, SQS_APPROXIMATE_RECEIVE_COUNT_LABEL (all string-valued; see taskiq_aio_sqs.message_metadata for the underlying SQSMessageMetadata dataclass).

This is implemented by wrapping the message body, in-memory, in a small JSON envelope between receiving it from SQS and handing it off to taskiq's receiver, it never changes what's actually stored in SQS, so it's safe to enable/disable at any time.

If you consume from broker.listen() directly (bypassing taskiq's Receiver), this metadata is always available on message.ack.metadata as an SQSMessageMetadata instance, regardless of the expose_message_metadata flag.

Heartbeat:

SQS's VisibilityTimeout hides a message from other consumers for a fixed window after it's received. If a task takes longer to run than that window, SQS assumes the original consumer died and redelivers the message, which with taskiq, means another worker (or the same one) picks it up and runs the task again, potentially publishing duplicate side effects (e.g. duplicate webhooks). taskiq-core's Receiver has no concept of "still working on it"; it only tells the broker about completion, via ack(), once the task and result backend are done. There's no built-in signal to periodically tell SQS "still alive" for a task that's still running.

Set enable_heartbeat=True to have the broker do this for you: while a message is being processed, a background task periodically calls ChangeMessageVisibility to push out the message's invisibility window, and is cancelled as soon as the message is acked.

broker = SQSBroker(
    sqs_queue_name="my-queue",
    enable_heartbeat=True,
    heartbeat_interval=15,
)

Notes:

• visibility_timeout doesn't need to be set explicitly: if it isn't (on the broker or on the relevant SQSQueue, for multi-queue configurations), the broker fetches it from SQS via GetQueueAttributes once at startup() and caches it for heartbeating. Setting it explicitly avoids that extra startup call.
• heartbeat_interval controls how often the visibility timeout is extended, in seconds. Defaults to half of the queue's visibility_timeout (explicit or fetched) if not set.
• heartbeat_max_extensions (default 100) caps how many times a single message's visibility timeout can be extended, as a safety net in case a message is never acked (e.g. a malformed message, an unregistered task name, or a pre_execute middleware exception), without this cap such a message would never become visible again for reprocessing or moving to a dead-letter queue.

Configuration:

SQS Broker parameters:

• endpoint_url - url to access sqs, this is not required, but is useful when running on local AWS emulators like MiniStack or LocalStack.
• sqs_queue_name - name of the sqs queue.
• region_name - region name, defaults to us-east-1.
• aws_access_key_id - aws access key id (Optional).
• aws_secret_access_key - aws secret access key (Optional).
• use_task_id_for_deduplication - use task_id for deduplication, this is useful when using a Fifo queue without content based deduplication, defaults to False.
• wait_time_seconds - wait time in seconds for long polling, defaults to 10.
• max_number_of_messages - maximum number of messages to receive, defaults to 1 (max 10).
• visibility_timeout - optional visibility timeout for received messages.
• delay_seconds - default delay for message delivery (0-900), defaults to 0.
• enable_batching - enable message batching for improved performance, defaults to False.
• batch_size - maximum number of messages to batch together (1-10), defaults to 10.
• batch_timeout - maximum time in seconds to wait before sending a partial batch (≥0.1), defaults to 1.0.
• skip_batch_tasks - list of task names that should bypass batching, defaults to None.
• s3_extended_bucket_name - extended bucket name for the s3 objects, adding this will allow the broker to kick messages that are too large for SQS by using S3 as well, by default the listen function handles this behaviour, defaults to None.
• task_id_generator - custom task_id generator (Optional).
• result_backend - custom result backend (Optional).
• is_fair_queue - : Whether the queue is a fair queue, if True, it will use the task_name as the MessageGroupId for all messages.
• expose_message_metadata - surface SQS delivery metadata (receipt handle, queue URL, message id, approximate receive count) to task code as TaskiqMessage labels, defaults to False. See Message Metadata.
• enable_heartbeat - periodically extend a message's visibility timeout in the background while it's being processed, to prevent premature redelivery for long-running tasks. If visibility_timeout isn't set explicitly, it's fetched from SQS at startup. Defaults to False. See Heartbeat.
• heartbeat_interval - how often, in seconds, to extend the visibility timeout while enable_heartbeat is True. Defaults to half of the relevant queue's visibility_timeout (explicit or fetched).
• heartbeat_max_extensions - safety cap on how many times a single message's visibility timeout may be extended, defaults to 100.

SQSQueue parameters (for multi-queue configuration):

• name - queue name.
• is_fifo - whether queue is FIFO.
• max_number_of_messages - per-queue receive batch size (1-10).
• wait_time_seconds - per-queue long poll wait (0-20).
• visibility_timeout - per-queue visibility timeout.
• options - reserved mapping for queue-level options.

Task Labels:

• delay - override the default delay for a specific task (0-900 seconds). Not supported for FIFO queues.
• group_id - set a custom MessageGroupId for FIFO queues or fair queues. Must be 1-128 characters, alphanumeric and specific punctuation only.
• skip_batching - bypass batching for a specific task call, set to True to send immediately.

S3 Result Backend parameters:

• bucket_name - name of the s3 bucket.
• base_path - base path for the s3 objects, defaults to "".
• endpoint_url - url to access s3, this is not required, but is useful when running on local AWS emulators like MiniStack or LocalStack.
• region_name - region name, defaults to us-east-1.
• aws_access_key_id - aws access key id (Optional).
• aws_secret_access_key - aws secret access key (Optional).
• serializer - custom serializer, defaults to OrjsonSerializer.

Local Development:

We use make to handle the commands for the project, you can see the available commands by running this in the root directory:

make

Setup

To setup the project, you can run the following commands:

make install

This installs the required dependencies for local development. The Makefile uses uv under the hood.

Linting

We use pre-commit to do linting locally, this will be included in the dev dependencies. We use ruff for linting and formatting, and pyright for static type checking. To install the pre-commit hooks, you can run the following command:

pre-commit install

If you for some reason hate pre-commit, you can run the following command to lint the code:

make check

Testing

To run tests, you can use the following command:

make test

In the background this will setup MiniStack (free open-source AWS emulator) to replicate the AWS services, and run the tests. It will also generate the coverage report and the badge.

You can also use Docker Compose to run MiniStack separately:

docker-compose up -d