feat sqs broker

.github/workflows/pr_tests.yaml

@@ -591,6 +591,72 @@ jobs:
           if-no-files-found: error
           include-hidden-files: true
 
+  test-sqs-smoke:
+    if: github.event.pull_request.draft == false
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          persist-credentials: false
+      - uses: astral-sh/setup-uv@08807647e7069bb48b6ef5acd8ec9567f424441b # v8.1.0
+        with:
+          version: "latest"
+      - name: Set up Python
+        uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
+        with:
+          python-version: "3.13"
+      - name: Install Dependencies
+        run: |
+          uv pip install --system --group optionals --group testing .
+      - name: Test
+        run: >
+          pytest -n auto
+          -vv -m "sqs and not connected"
+
+  test-sqs-real:
+    if: github.event.pull_request.draft == false
+    runs-on: ubuntu-latest
+    needs:
+      - test-basic
+      - test-sqs-smoke
+    services:
+      sqs:
+        image: softwaremill/elasticmq-native:1.7.1
+        ports:
+          - 9324:9324
+        options: >-
+          --health-cmd "wget --no-verbose --tries=1 --spider http://localhost:9324/health"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 20
+    steps:
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          persist-credentials: false
+      - uses: astral-sh/setup-uv@08807647e7069bb48b6ef5acd8ec9567f424441b # v8.1.0
+        with:
+          version: "latest"
+      - name: Set up Python
+        uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
+        with:
+          python-version: "3.13"
+      - name: Install Dependencies
+        run: |
+          uv pip install --system --group optionals --group testing .
+      - name: Test
+        run: >
+          pytest --cov --cov-report=
+          -vv -m "(slow and sqs and connected) or (sqs and connected)"
+      - name: Rename coverage file
+        run: mkdir coverage && mv .coverage coverage/.coverage.sqs-py
+      - name: Store coverage files
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: .coverage.sqs-py
+          path: coverage
+          if-no-files-found: error
+          include-hidden-files: true
+
   coverage-combine:
     if: github.event.pull_request.draft == false
     needs:
@@ -604,6 +670,7 @@ jobs:
       - test-redis-real
       - test-redis-cluster-real
       - test-mqtt-real
+      - test-sqs-real
     runs-on: ubuntu-latest
 
     steps:

docker-compose.yaml

@@ -80,6 +80,20 @@ services:
     security_opt:
       - no-new-privileges:true
 
+  sqs:
+    image: softwaremill/elasticmq-native:1.7.1
+    ports:
+      - "9324:9324"
+      - "9325:9325"
+    healthcheck:
+      test: ["CMD", "wget", "--no-verbose", "--tries=1", "--spider", "http://localhost:9324/health"]
+      interval: 5s
+      timeout: 5s
+      retries: 20
+      start_period: 10s
+    security_opt:
+      - no-new-privileges:true
+
   faststream:
     build: .
     volumes:

docs/docs/en/sqs/ack.md

@@ -0,0 +1,32 @@
+# Acknowledgement
+
+SQS acknowledgement maps onto the queue API:
+
+| FastStream | SQS action | Effect |
+|------------|-----------|--------|
+| `ack` | `DeleteMessage` | Message handled — removed from the queue. |
+| `nack` | `ChangeMessageVisibility(0)` | Returned immediately for redelivery. |
+| `reject` | `DeleteMessage` | Dropped (route to a DLQ via a redrive policy). |
+
+The default [`AckPolicy`](../getting-started/acknowledgement.md) for SQS
+subscribers is `ACK` (the message is deleted after the handler succeeds, giving
+at-least-once delivery). On error the message is **not** deleted and SQS
+redelivers it once its visibility timeout expires.
+
+```python linenums="1"
+{! docs_src/sqs/ack/policy.py [ln:3,6] !}
+```
+
+## Manual acknowledgement
+
+```python linenums="1"
+{! docs_src/sqs/ack/manual.py [ln:5,11-14] !}
+```
+
+## Dead-letter queues
+
+Route exhausted messages to a DLQ with a redrive policy on the queue:
+
+```python linenums="1"
+{! docs_src/sqs/ack/dlq.py [ln:4,9-15] !}
+```

docs/docs/en/sqs/fifo.md

@@ -0,0 +1,34 @@
+# FIFO queues
+
+FIFO queues guarantee ordering and exactly-once processing within a message
+group. Declare one with `FifoQueue` — the `.fifo` suffix is added automatically:
+
+```python linenums="1"
+{! docs_src/sqs/fifo/app.py [ln:3-18] !}
+```
+
+## Publishing to FIFO queues
+
+FIFO sends require a `MessageGroupId`. When content-based deduplication is off,
+also provide a `deduplication_id`:
+
+```python linenums="1"
+{! docs_src/sqs/fifo/app.py [ln:22-27] !}
+```
+
+Messages sharing a `group_id` are delivered in strict order; different groups
+are processed in parallel.
+
+## Receive-retry deduplication (`request_attempt_id`)
+
+For FIFO queues you can pass a `request_attempt_id` to the subscriber. It maps to
+the SQS `ReceiveRequestAttemptId` — the deduplication token SQS uses when a
+`ReceiveMessage` call fails (e.g. a network error): retrying with the same token
+returns the *same* batch of messages instead of leaving them invisible.
+
+```python linenums="1"
+{! docs_src/sqs/fifo/request_attempt_id.py [ln:3-13] !}
+```
+
+This parameter is FIFO-only — setting it on a standard (non-`.fifo`) queue raises
+a `SetupError` at startup.

docs/docs/en/sqs/index.md

@@ -0,0 +1,50 @@
+---
+# 0.5 - API
+# 2 - Release
+# 3 - Contributing
+# 5 - Template Page
+# 10 - Default
+search:
+  boost: 10
+---
+
+# AWS SQS routing
+
+!!! note ""
+    **FastStream** SQS support is implemented on top of [**aiobotocore**](https://pypi.org/project/aiobotocore/){.external-link target="_blank"} — an `asyncio` wrapper around `botocore`. You can reach the underlying `aiobotocore` SQS client through `broker.config.client` when you need APIs not wrapped by FastStream.
+
+## Why SQS
+
+[Amazon SQS](https://aws.amazon.com/sqs/){.external-link target="_blank"} is a fully managed message queue. Unlike push-based brokers, SQS is **pull-based**: consumers long-poll a queue for messages, process them, and explicitly delete them when done. SQS offers **standard** (at-least-once, best-effort ordering) and **FIFO** (exactly-once processing, strict ordering) queues.
+
+## FastStream `SQSBroker`
+
+```python linenums="1"
+{! docs_src/sqs/index/basic.py [ln:3-17] !}
+```
+
+### Connection parameters
+
+`SQSBroker` accepts the same connection options as `aiobotocore`'s `create_client`:
+
+| Parameter | Description |
+|-----------|-------------|
+| `region_name` | AWS region, e.g. `"us-east-1"`. |
+| `endpoint_url` | Custom endpoint — point this at **LocalStack** (`http://localhost:4566`) or ElasticMQ for local development. |
+| `aws_access_key_id` / `aws_secret_access_key` / `aws_session_token` | Explicit credentials (otherwise resolved from the environment / `~/.aws`). |
+| `use_ssl`, `verify`, `botocore_config` | Standard botocore client tuning. |
+| `response_queue` | Queue used for [RPC replies](rpc.md). |
+
+### Local development with LocalStack
+
+```python linenums="1"
+{! docs_src/sqs/index/localstack.py [ln:6-11] !}
+```
+
+## Testing
+
+Use `TestSQSBroker` to route messages in memory — no AWS connection required:
+
+```python linenums="1"
+{! docs_src/sqs/index/testing.py !}
+```

docs/docs/en/sqs/publishing.md

@@ -0,0 +1,52 @@
+# Publishing
+
+Publish to a queue with `broker.publish`:
+
+```python linenums="1"
+{! docs_src/sqs/publishing/publish.py [ln:16-21] !}
+```
+
+Headers are sent as SQS **MessageAttributes**. FastStream reserves the
+`content-type`, `reply_to`, `correlation_id`, `empty-body`, and `base64-body`
+attribute names for transport metadata.
+
+!!! note "Binary payloads"
+    SQS accepts only text message bodies. When you publish `bytes` that are not
+    valid UTF-8 (an image, gzip, protobuf, ...), FastStream sends them
+    base64-encoded and sets the reserved `base64-body` message attribute, so a
+    FastStream subscriber receives the original bytes back transparently.
+    Non-FastStream consumers of such messages should base64-decode the body when
+    that attribute is present.
+
+## Declared queues
+
+Pass an `SQSQueue` instead of a name to have FastStream create the queue (with
+the given attributes) on startup:
+
+```python linenums="1"
+{! docs_src/sqs/publishing/declared_queue.py [ln:4,9,18] !}
+```
+
+## Publisher objects
+
+Register a reusable publisher with `@broker.publisher`:
+
+```python linenums="1"
+{! docs_src/sqs/publishing/publisher_object.py [ln:9,12-15] !}
+```
+
+## Batch publishing
+
+```python linenums="1"
+{! docs_src/sqs/publishing/batch.py [ln:16] !}
+```
+
+A single `SendMessageBatch` request carries up to 10 messages; larger batches
+are split into several requests automatically.
+
+You can also declare a reusable batch publisher — every `publish(*messages)`
+call becomes one `SendMessageBatch` request:
+
+```python linenums="1"
+{! docs_src/sqs/publishing/batch_publisher.py [ln:9,16-17] !}
+```

docs/docs/en/sqs/rpc.md

@@ -0,0 +1,21 @@
+# Request / response
+
+SQS has no native request/reply, so FastStream implements RPC with a dedicated
+**response queue** plus a `correlation_id`. Configure the response queue on the
+broker, then call `broker.request`:
+
+```python linenums="1"
+{! docs_src/sqs/rpc/app.py [ln:3-17] !}
+```
+
+How it works:
+
+1. The producer sends the request with `reply_to` set to the response queue URL
+   and a generated `correlation_id`.
+2. A background task long-polls the response queue, matching replies to pending
+   requests by `correlation_id`, and deletes them once consumed.
+3. The handler's return value is published back to `reply_to`.
+
+!!! note ""
+    `broker.request` raises `FeatureNotSupportedException` if no `response_queue`
+    was configured.

docs/docs/en/sqs/security.md

@@ -0,0 +1,22 @@
+# Security
+
+Credentials are usually resolved by `botocore` from the environment, an IAM
+role, or `~/.aws/credentials`. You can also pass them explicitly or via a
+FastStream security object.
+
+## Explicit credentials
+
+```python linenums="1"
+{! docs_src/sqs/security/explicit.py [ln:6-11] !}
+```
+
+## Security objects
+
+`SASLPlaintext` maps `username`/`password` onto the access key id / secret, and
+`BaseSecurity` controls TLS:
+
+```python linenums="1"
+{! docs_src/sqs/security/sasl.py [ln:4,7-14] !}
+```
+
+The configured security also drives the generated AsyncAPI schema.

docs/docs/en/sqs/subscription.md

@@ -0,0 +1,57 @@
+# Subscription
+
+Each subscriber long-polls its queue in a background task and dispatches every
+received message to your handler:
+
+```python linenums="1"
+{! docs_src/sqs/subscription/basic.py [ln:10-16] !}
+```
+
+| Parameter | Description |
+|-----------|-------------|
+| `wait_time_seconds` | SQS `WaitTimeSeconds` for long polling. Higher values reduce empty receives and request cost. |
+| `max_messages` | SQS `MaxNumberOfMessages` per receive. |
+| `visibility_timeout` | How long a received message stays invisible to other consumers while being processed. |
+
+## Accessing message metadata
+
+```python linenums="1"
+{! docs_src/sqs/subscription/message_info.py [ln:5,11-13] !}
+```
+
+## Declaring queues on subscribe
+
+Pass an `SQSQueue`/`FifoQueue` to create the queue automatically on startup:
+
+```python linenums="1"
+{! docs_src/sqs/subscription/declared_queue.py [ln:4,10-11] !}
+```
+
+## Concurrent processing
+
+By default messages from one poll are processed sequentially. Set `max_workers`
+to handle up to that many messages at the same time:
+
+```python linenums="1"
+{! docs_src/sqs/subscription/concurrency.py [ln:10-11] !}
+```
+
+!!! warning
+    `max_workers` can't be combined with `batch=True` or FIFO queues —
+    concurrent processing would break message-group ordering.
+
+## Long-running handlers
+
+If a handler runs longer than the queue's `VisibilityTimeout`, SQS redelivers
+the message while the first consumer is still working. Enable
+`extend_visibility` to keep extending the timeout with a background heartbeat
+until the handler finishes:
+
+```python linenums="1"
+{! docs_src/sqs/subscription/long_running.py [ln:10-15] !}
+```
+
+!!! note
+    `extend_visibility=True` requires an explicit `visibility_timeout` — the
+    heartbeat re-extends the message by that value every `visibility_timeout / 2`
+    seconds.

docs/docs/navigation_template.txt

@@ -99,6 +99,13 @@ search:
     - [Shared subscriptions](mqtt/shared.md)
     - [MQTT 3.1.1 vs 5.0](mqtt/versions.md)
     - [Security](mqtt/security.md)
+- [AWS SQS](sqs/index.md)
+    - [Publishing](sqs/publishing.md)
+    - [Subscription](sqs/subscription.md)
+    - [FIFO queues](sqs/fifo.md)
+    - [Request / response](sqs/rpc.md)
+    - [Acknowledgement](sqs/ack.md)
+    - [Security](sqs/security.md)
 - [NATS](nats/index.md)
     - Subscription
         - [Direct](nats/examples/direct.md)