AWS SQS vs SNS Explained — Queues, Topics and When to Use Each

import boto3
import json
import time

# --- PRODUCER: Order Service drops a new order into the queue ---

sqs_client = boto3.client('sqs', region_name='us-east-1')

ORDER_QUEUE_URL = 'https://sqs.us-east-1.amazonaws.com/123456789012/order-processing-queue'

def place_order(order_id: str, customer_email: str, items: list) -> dict:
    """
    Sends a new order message to SQS.
    The fulfilment service will pick this up independently.
    """
    order_payload = {
        'order_id': order_id,
        'customer_email': customer_email,
        'items': items,
        'timestamp': time.time()
    }

    # MessageGroupId is only needed for FIFO queues.
    # For Standard queues, just send the body.
    response = sqs_client.send_message(
        QueueUrl=ORDER_QUEUE_URL,
        MessageBody=json.dumps(order_payload),
        # DelaySeconds postpones visibility to consumers — useful for
        # scheduling near-future jobs (max 900 seconds = 15 minutes)
        DelaySeconds=0
    )

    print(f"[PRODUCER] Message sent. SQS Message ID: {response['MessageId']}")
    return response


# --- CONSUMER: Fulfilment Service polls and processes orders ---

def process_pending_orders():
    """
    Polls the queue for up to 10 messages at a time.
    Uses long polling (WaitTimeSeconds=20) to reduce empty responses
    and lower your AWS bill — this is the single most impactful SQS setting.
    """
    print("[CONSUMER] Polling for messages...")

    response = sqs_client.receive_message(
        QueueUrl=ORDER_QUEUE_URL,
        MaxNumberOfMessages=10,   # Batch up to 10 messages per poll
        WaitTimeSeconds=20,       # Long polling — ALWAYS use this
        VisibilityTimeout=60      # Give our worker 60s to finish before retry
    )

    messages = response.get('Messages', [])

    if not messages:
        print("[CONSUMER] Queue is empty. Nothing to process.")
        return

    for message in messages:
        receipt_handle = message['ReceiptHandle']  # Needed to delete the message
        order = json.loads(message['Body'])

        print(f"[CONSUMER] Processing order {order['order_id']} "
              f"for {order['customer_email']}")

        # Simulate doing the actual work (charge card, update DB, etc.)
        fulfil_order(order)

        # CRITICAL: Delete the message after successful processing.
        # If you skip this, SQS re-delivers it after VisibilityTimeout expires.
        sqs_client.delete_message(
            QueueUrl=ORDER_QUEUE_URL,
            ReceiptHandle=receipt_handle
        )
        print(f"[CONSUMER] Order {order['order_id']} done. Message deleted.")


def fulfil_order(order: dict):
    """Placeholder for real business logic."""
    print(f"  -> Charging card and packing {len(order['items'])} item(s)...")
    time.sleep(0.1)  # Simulate processing time


# --- Run it ---
place_order(
    order_id='ORD-9981',
    customer_email='alex@example.com',
    items=['Laptop Stand', 'USB-C Hub']
)

process_pending_orders()

import boto3
import json

sns_client = boto3.client('sns', region_name='us-east-1')

USER_EVENTS_TOPIC_ARN = 'arn:aws:sns:us-east-1:123456789012:user-lifecycle-events'

def publish_user_registered_event(user_id: str, email: str, plan: str, region: str):
    """
    Publishes a single UserRegistered event to the SNS topic.
    SNS delivers this simultaneously to ALL subscribers:
      - SQS queue -> welcome-email-service
      - SQS queue -> crm-sync-service (filter: plan=pro)
      - Lambda   -> real-time-analytics
    The auth service doesn't know or care who those subscribers are.
    """
    event_payload = {
        'event_type': 'UserRegistered',
        'user_id': user_id,
        'email': email,
        'plan': plan,
        'region': region
    }

    response = sns_client.publish(
        TopicArn=USER_EVENTS_TOPIC_ARN,
        Message=json.dumps(event_payload),
        Subject='UserRegistered',  # Used by email subscribers as the email subject

        # MessageAttributes enable server-side filtering.
        # A subscription with filter {"plan": ["pro"]} only receives pro-plan signups.
        # This avoids every subscriber having to filter in their own code.
        MessageAttributes={
            'event_type': {
                'DataType': 'String',
                'StringValue': 'UserRegistered'
            },
            'plan': {
                'DataType': 'String',
                'StringValue': plan  # 'free' or 'pro'
            },
            'region': {
                'DataType': 'String',
                'StringValue': region  # 'EU', 'US', 'APAC'
            }
        }
    )

    print(f"[SNS] Event published. Message ID: {response['MessageId']}")
    print(f"[SNS] Delivered to all active subscribers in parallel.")
    return response


# --- Subscribing an SQS queue to this SNS topic (infra setup, done once) ---

def subscribe_sqs_queue_to_topic(
    topic_arn: str,
    queue_arn: str,
    filter_policy: dict = None
) -> str:
    """
    Wires an SQS queue as an SNS subscriber.
    Optionally attach a filter so this queue only gets relevant messages.
    """
    subscribe_kwargs = {
        'TopicArn': topic_arn,
        'Protocol': 'sqs',
        'Endpoint': queue_arn
    }

    if filter_policy:
        # FilterPolicy limits which messages reach this subscription.
        # The CRM service only wants to hear about 'pro' plan signups.
        subscribe_kwargs['Attributes'] = {
            'FilterPolicy': json.dumps(filter_policy)
        }

    response = sns_client.subscribe(**subscribe_kwargs)
    print(f"[SNS] Queue subscribed. Subscription ARN: {response['SubscriptionArn']}")
    return response['SubscriptionArn']


# --- Example: set up the CRM subscription with a filter ---
subscribe_sqs_queue_to_topic(
    topic_arn=USER_EVENTS_TOPIC_ARN,
    queue_arn='arn:aws:sqs:us-east-1:123456789012:crm-sync-queue',
    filter_policy={'plan': ['pro']}  # Only pro signups reach the CRM queue
)

# --- Publish a new user signup ---
publish_user_registered_event(
    user_id='usr-4471',
    email='jordan@example.com',
    plan='pro',
    region='EU'
)

import boto3
import json

sqs_client = boto3.client('sqs', region_name='us-east-1')
sns_client = boto3.client('sns', region_name='us-east-1')


def create_queue_with_dlq(main_queue_name: str, dlq_name: str) -> tuple[str, str]:
    """
    Creates a main SQS queue and wires up a Dead-Letter Queue.
    If a message fails maxReceiveCount times, it moves to the DLQ
    instead of looping forever or vanishing silently.
    """

    # Step 1: Create the DLQ first — it's just a regular SQS queue
    dlq_response = sqs_client.create_queue(
        QueueName=dlq_name,
        Attributes={
            'MessageRetentionPeriod': '1209600'  # Keep failed messages for 14 days
        }
    )
    dlq_url = dlq_response['QueueUrl']
    print(f"[SETUP] DLQ created: {dlq_url}")

    # Step 2: Get the DLQ's ARN — needed to reference it as a redrive target
    dlq_attributes = sqs_client.get_queue_attributes(
        QueueUrl=dlq_url,
        AttributeNames=['QueueArn']
    )
    dlq_arn = dlq_attributes['Attributes']['QueueArn']

    # Step 3: Create the main queue with a RedrivePolicy pointing at the DLQ
    # maxReceiveCount=3 means: after 3 failed delivery attempts, move to DLQ
    redrive_policy = {
        'deadLetterTargetArn': dlq_arn,
        'maxReceiveCount': '3'
    }

    main_queue_response = sqs_client.create_queue(
        QueueName=main_queue_name,
        Attributes={
            'VisibilityTimeout': '60',
            'RedrivePolicy': json.dumps(redrive_policy)
        }
    )
    main_queue_url = main_queue_response['QueueUrl']
    print(f"[SETUP] Main queue created with DLQ attached: {main_queue_url}")

    return main_queue_url, dlq_url


def wire_sns_to_sqs_fanout(
    topic_arn: str,
    queue_url: str,
    queue_arn: str
):
    """
    Subscribes an SQS queue to an SNS topic — the fan-out pattern.
    Also grants SNS permission to write to the SQS queue.
    Forgetting the SQS access policy is the #1 cause of 'messages not arriving' bugs.
    """

    # Step 1: Grant SNS permission to send messages to this SQS queue
    # Without this policy, SNS silently drops messages — no error thrown
    sqs_policy = {
        'Version': '2012-10-17',
        'Statement': [{
            'Sid': 'AllowSNSToSendToSQS',
            'Effect': 'Allow',
            'Principal': {'Service': 'sns.amazonaws.com'},
            'Action': 'sqs:SendMessage',
            'Resource': queue_arn,
            'Condition': {
                'ArnEquals': {'aws:SourceArn': topic_arn}
            }
        }]
    }

    sqs_client.set_queue_attributes(
        QueueUrl=queue_url,
        Attributes={'Policy': json.dumps(sqs_policy)}
    )
    print(f"[SETUP] SQS access policy updated to allow SNS.")

    # Step 2: Subscribe the queue to the topic
    # RawMessageDelivery=true means the SQS message body IS your JSON payload
    # Without it, the body is wrapped in an SNS envelope — harder to parse
    subscription = sns_client.subscribe(
        TopicArn=topic_arn,
        Protocol='sqs',
        Endpoint=queue_arn,
        Attributes={
            'RawMessageDelivery': 'true'  # Skip the SNS wrapper — always set this
        }
    )

    print(f"[SETUP] Fan-out wired. Subscription ARN: {subscription['SubscriptionArn']}")
    return subscription['SubscriptionArn']


# --- Run the setup ---
main_url, dlq_url = create_queue_with_dlq(
    main_queue_name='order-fulfilment-queue',
    dlq_name='order-fulfilment-dlq'
)

wire_sns_to_sqs_fanout(
    topic_arn='arn:aws:sns:us-east-1:123456789012:order-events',
    queue_url=main_url,
    queue_arn='arn:aws:sqs:us-east-1:123456789012:order-fulfilment-queue'
)

print("\n[READY] Production fan-out pattern active:")
print("  SNS Topic -> SQS Queue (durable) -> Lambda/Worker")
print("  Failed messages -> DLQ after 3 attempts")