SQS & SNS
SQS provides durable queues; SNS provides fan-out pub/sub. Python workers use boto3 to publish events, poll queues, and build decoupled pipelines between services and Lambda functions.
Recipe
import boto3
sqs = boto3.client("sqs")
queue_url = sqs.get_queue_url(QueueName="orders")["QueueUrl"]
sqs.send_message(QueueUrl=queue_url, MessageBody='{"order_id": "1"}')When to reach for this:
- Decouple producers and consumers with backpressure via queue depth
- Fan-out notifications to many subscribers with SNS
- Retry and DLQ patterns for failed processing
- Bridge to Lambda without custom polling infrastructure
Working Example
Publish to SNS, subscribe SQS, send/receive/delete with long polling.
import json
import boto3
sns = boto3.client("sns")
sqs = boto3.client("sqs")
TOPIC_ARN = "arn:aws:sns:us-east-1:123456789012:orders-events"
QUEUE_URL = sqs.get_queue_url(QueueName="orders-worker")["QueueUrl"]
def publish_order_event(order_id: str) -> None:
sns.publish(
TopicArn=TOPIC_ARN,
Message=json.dumps({"order_id": order_id, "type": "created"}),
)
def send_to_queue(body: dict) -> None:
sqs.send_message(QueueUrl=QUEUE_URL, MessageBody=json.dumps(body))
def receive_batch(max_messages: int = 5) -> list[dict]:
resp = sqs.receive_message(
QueueUrl=QUEUE_URL,
MaxNumberOfMessages=max_messages,
WaitTimeSeconds=20,
MessageAttributeNames=["All"],
)
return resp.get("Messages", [])
def delete_message(receipt_handle: str) -> None:
sqs.delete_message(QueueUrl=QUEUE_URL, ReceiptHandle=receipt_handle)
if __name__ == "__main__":
send_to_queue({"order_id": "42"})
for msg in receive_batch():
print(msg["Body"])
delete_message(msg["ReceiptHandle"])What this demonstrates:
- SNS
publishfor fan-out; SQSsend_messagefor point-to-point - Long polling (
WaitTimeSeconds=20) reduces empty receives and cost - Delete only after successful processing - otherwise message returns after visibility timeout
Deep Dive
How It Works
- SQS standard queues are at-least-once; FIFO queues add ordering and deduplication
- Visibility timeout hides message during processing; extend with
change_message_visibilityfor long jobs - SNS delivers to SQS, Lambda, HTTP, email subscribers based on subscription policy
- DLQ captures poison messages after max receive count
Patterns
| Pattern | Service |
|---|---|
| Task queue | SQS + worker |
| Broadcast | SNS topic |
| SNS → SQS fan-out | Multiple queues subscribed |
Python Notes
# Partial batch failure reporting for Lambda SQS event source (conceptual)
def handler(event, context):
failures = []
for record in event["Records"]:
try:
process(json.loads(record["body"]))
except Exception:
failures.append({"itemIdentifier": record["messageId"]})
return {"batchItemFailures": failures}Gotchas
- Deleting before processing completes - data loss. Fix: delete after commit/side effects succeed.
- Visibility timeout too short - duplicate processing. Fix: tune timeout > p99 handler duration.
- No DLQ - poison messages loop forever. Fix: redrive policy to DLQ with alarm on depth.
- Publishing huge payloads - 256KB SQS limit. Fix: store blob in S3, pass pointer in message.
- Polling without long wait - CPU burn and higher costs. Fix:
WaitTimeSeconds10-20.
Alternatives
| Alternative | Use When | Don't Use When |
|---|---|---|
| Kafka/MSK | High throughput, replay log | Simple AWS-native messaging |
| Celery + Redis | Existing Python task queue | Want fully managed scaling |
| EventBridge | Event routing with rules | Simple single-queue worker |
FAQs
Standard vs FIFO?
FIFO when order and exactly-once processing matter per message group; standard for highest throughput tolerant of duplicates.
How do I make processing idempotent?
Use dedupe id in DynamoDB or DB unique constraint on business key.
SNS vs SQS?
SNS pushes to many subscribers; SQS buffers for workers that pull at their own rate.
How do I monitor queues?
CloudWatch ApproximateNumberOfMessagesVisible and age of oldest message alarms.
Can boto3 create queues?
create_queue in bootstrap scripts or IaC; apps usually only need URLs/ARNs from config.
What about message attributes?
Use for metadata (content-type, trace id) without parsing body JSON.
How do I test?
moto mocks send/receive; LocalStack for integration; always test visibility timeout behavior in staging.
How many messages per receive?
Up to 10 with MaxNumberOfMessages - batch processing amortizes API calls.
How does SNS filter to SQS?
Subscription filter policies on message JSON attributes reduce noise to each queue.
How do I encrypt messages?
Enable SSE on queues/topics; use KMS CMK for compliance requirements.
Related
- Lambda - SQS event source mapping
- boto3 Basics - client reuse in workers
- Retry, Pagination & Throttling - API retries
- S3 - large payload pointers
- Cloud SDK Best Practices - worker patterns
Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.