AWS SQS

AWS SQS (Simple Queue Service) is AWS's managed message queue service.

Installation

pip install onestep-sqs

Authentication Configuration

Environment Variables (Recommended)

export AWS_ACCESS_KEY_ID=your_access_key
export AWS_SECRET_ACCESS_KEY=your_secret_key
export AWS_DEFAULT_REGION=us-east-1

Or Use IAM Role (EC2/Lambda)

When running on EC2 or Lambda, IAM Role authentication is used automatically - no need to configure keys.

Basic Usage

from onestep import OneStepApp
from onestep_sqs import SQSConnector

app = OneStepApp("sqs-demo")

# Create connector
sqs = SQSConnector(region_name="us-east-1")

# Create queue Source
source = sqs.queue(
    "https://sqs.us-east-1.amazonaws.com/123456789012/my-queue",
    batch_size=10,
)

# Create queue Sink
sink = sqs.queue(
    "https://sqs.us-east-1.amazonaws.com/123456789012/results-queue",
)


@app.task(source=source, emit=sink, concurrency=8)
async def process_message(ctx, item):
    print(f"Processing message: {item}")
    return {"result": "done"}


if __name__ == "__main__":
    app.run()

Queue Configuration

Standard Queue

sqs = SQSConnector(region_name="us-east-1")
source = sqs.queue(
    "https://sqs.us-east-1.amazonaws.com/123456789012/standard-queue"
)

FIFO Queue

source = sqs.queue(
    "https://sqs.us-east-1.amazonaws.com/123456789012/my-queue.fifo",
    message_group_id="default-group",  # Required for FIFO
)

Advanced Configuration

source = sqs.queue(
    "https://sqs.../my-queue",
    batch_size=10,                  # Messages per poll
    wait_time_s=20,                 # Long poll seconds
    delete_batch_size=10,           # Batch delete count
    delete_flush_interval_s=0.5,    # Batch delete interval
    heartbeat_interval_s=15,        # Heartbeat interval
    heartbeat_visibility_timeout=60, # Visibility timeout
)

Publishing Messages

Publish via Sink

@app.task(source=..., emit=sink)
async def process(ctx, item):
    return {"result": "data"}  # Auto-published to sink

Manual Publish

import asyncio

async def main():
    sink = sqs.queue("https://sqs.../my-queue")
    
    # Publish single
    await sink.publish({"job": "data"})
    
    # Publish multiple
    for i in range(100):
        await sink.publish({"id": i})

asyncio.run(main())

FIFO Message Grouping

sink = sqs.queue(
    "https://sqs.../my-queue.fifo",
    message_group_id="group-1",
)

# Group by user ID
async def publish_for_user(user_id, data):
    sink = sqs.queue(
        "https://sqs.../my-queue.fifo",
        message_group_id=f"user-{user_id}",
    )
    await sink.publish(data)

Visibility Timeout

After a message is consumed, it becomes invisible to other consumers until the visibility timeout expires:

source = sqs.queue(
    "https://sqs.../my-queue",
    heartbeat_interval_s=15,        # Renew every 15 seconds
    heartbeat_visibility_timeout=60, # Renew to 60 seconds
)


@app.task(source=source)
async def long_task(ctx, item):
    await asyncio.sleep(45)  # Long task, visibility auto-renewed

Dead Letter Queue

Configure SQS dead letter queue:

# Configure the dead letter queue in AWS Console or CloudFormation
# The main queue's Redrive Policy points to the dead letter queue

# Handle dead letters in onestep
dead_letter = sqs.queue("https://sqs.../dead-letter-queue")


@app.task(source=dead_letter)
async def handle_dead_letter(ctx, item):
    print(f"Dead letter message: {item}")

YAML Configuration

resources:
  sqs:
    type: sqs
    region_name: "us-east-1"
  
  jobs:
    type: sqs_queue
    connector: sqs
    url: "https://sqs.us-east-1.amazonaws.com/123456789012/jobs"
  
  results:
    type: sqs_queue
    connector: sqs
    url: "https://sqs.us-east-1.amazonaws.com/123456789012/results"

tasks:
  - name: process_jobs
    source: jobs
    emit: results
    concurrency: 8

Best Practices

1. Use IAM Role

Use IAM Role on EC2/Lambda to avoid hard-coding keys:

# Automatically uses the instance's IAM Role
sqs = SQSConnector(region_name="us-east-1")

2. Batch Operations

# Adjust batch parameters for higher throughput
source = sqs.queue(
    "https://sqs.../my-queue",
    batch_size=10,
    delete_batch_size=10,
    delete_flush_interval_s=0.5,
)

3. Concurrency Control

# Adjust concurrency based on task processing time
@app.task(source=source, concurrency=16)
async def fast_task(ctx, item):
    ...

@app.task(source=source, concurrency=4)
async def slow_task(ctx, item):
    ...

4. Error Handling

from onestep import MaxAttempts

@app.task(
    source=source,
    retry=MaxAttempts(max_attempts=3, delay_s=5.0)
)
async def might_fail(ctx, item):
    ...

5. Monitoring

Use CloudWatch to monitor queues:

• ApproximateNumberOfMessagesVisible: Visible messages count
• ApproximateNumberOfMessagesNotVisible: Invisible messages count
• NumberOfMessagesSent: Sent messages count
• NumberOfMessagesReceived: Received messages count
• NumberOfMessagesDeleted: Deleted messages count
• NumberOfMessagesFailed: Failed messages count