Large payloads with S3

SQS messages are capped at 256 KiB, which real workloads can outgrow quickly — a document to summarize, a batch of rows to import, a rendered file to process. S3OffloadMiddleware transparently moves oversized payloads through S3 instead, and pairs naturally with S3ResultBackend for the result on the way back — both are just S3 buckets, configured the same way.

"""
Run worker:
    taskiq worker docs.examples.large_payloads_with_s3:broker

Run this script:
    python docs/examples/large_payloads_with_s3.py
"""

import asyncio

import dotenv

from taskiq_sqs import S3OffloadMiddleware, S3ResultBackend, SQSBroker
from taskiq_sqs.types import S3Bucket, SQSQueue


dotenv.load_dotenv()

ENDPOINT_URL = "http://localhost:4566"
AWS_REGION = "us-east-1"

broker = SQSBroker(
    queues=SQSQueue(name="large-payload-queue"),
    endpoint_url=ENDPOINT_URL,
    aws_region_name=AWS_REGION,
).with_result_backend(
    S3ResultBackend(
        bucket=S3Bucket(name="large-payload-results"),
        endpoint_url=ENDPOINT_URL,
        aws_region_name=AWS_REGION,
    ),
)
broker.add_middlewares(
    S3OffloadMiddleware(
        bucket=S3Bucket(name="large-payload-offload"),
        max_message_size=200_000,  # payloads larger than this many bytes are offloaded to S3
        endpoint_url=ENDPOINT_URL,
        aws_region_name=AWS_REGION,
    ),
)


@broker.task()
async def summarize_document(content: str) -> dict[str, int]:
    return {"characters": len(content), "words": len(content.split())}


async def main() -> None:
    await broker.startup()
    document = "taskiq-sqs " * 100_000  # well over the 256 KiB SQS message limit
    task = await summarize_document.kiq(document)
    result = await task.wait_result(timeout=10)
    print(result.return_value)
    await broker.shutdown()


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

This example uses three separate resources: the queue itself, a bucket for offloaded payloads, and a bucket for results. All three are declared automatically on startup(), same as in the basic example.

To run it:

Start a worker:

taskiq worker docs.examples.large_payloads_with_s3:broker

In another terminal, run the script. It builds an ~1.1 MB string — well past the SQS limit — kicks it, and waits for the result:
```
python docs/examples/large_payloads_with_s3.py
```

You should see something like {'characters': 1100000, 'words': 100000} printed once the worker finishes. Behind the scenes: the middleware uploaded the document to large-payload-offload before the message ever reached SQS, the worker downloaded it back before running summarize_document, deleted it from S3 afterwards (the default delete_after_execute=True), and the result itself was written to large-payload-results.