Amazon SQS #

pip install boto3

import boto3
import os

# ANTI-PATTERN: hardcoding credentials in code
sqs = boto3.client(
    "sqs",
    aws_access_key_id="«redacted:AKIA…»",      # ✗ -- never do this
    aws_secret_access_key="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
)

# CORRECT way 1: read from environment variables
sqs = boto3.client(
    "sqs",
    region_name=os.getenv("AWS_DEFAULT_REGION", "ap-southeast-1"),
    aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
    aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
    aws_session_token=os.getenv("AWS_SESSION_TOKEN")   # if using temporary credentials
)

# CORRECT way 2: boto3 automatically reads from ~/.aws/credentials or the IAM Role
sqs = boto3.client("sqs", region_name="ap-southeast-1")

# CORRECT way 3: use the resource API (more pythonic)
sqs_resource = boto3.resource("sqs", region_name="ap-southeast-1")

import boto3
import json
import os

sqs = boto3.client("sqs", region_name=os.getenv("AWS_DEFAULT_REGION", "ap-southeast-1"))

def buat_standard_queue(nama: str, retention_detik: int = 86400) -> str:
    """Create a Standard Queue, return the URL."""
    response = sqs.create_queue(
        QueueName=nama,
        Attributes={
            "VisibilityTimeout":      "30",              # seconds a message is hidden while processed
            "MessageRetentionPeriod": str(retention_detik),  # how long messages are kept (max 14 days)
            "ReceiveMessageWaitTimeSeconds": "20",       # Long Polling -- wait up to 20 seconds
            "RedrivePolicy": json.dumps({               # Dead Letter Queue after 3 failures
                "deadLetterTargetArn": buat_dlq(nama + "-dlq"),
                "maxReceiveCount":     "3"
            })
        }
    )
    print(f"Standard Queue created: {response['QueueUrl']}")
    return response["QueueUrl"]

def buat_fifo_queue(nama: str) -> str:
    """Create a FIFO Queue -- the name must end with .fifo"""
    if not nama.endswith(".fifo"):
        nama += ".fifo"

    response = sqs.create_queue(
        QueueName=nama,
        Attributes={
            "FifoQueue":                    "true",
            "ContentBasedDeduplication":    "true",   # auto-dedup based on message content
            "VisibilityTimeout":            "60",
            "MessageRetentionPeriod":       "86400",
            "ReceiveMessageWaitTimeSeconds": "20",
        }
    )
    print(f"FIFO Queue created: {response['QueueUrl']}")
    return response["QueueUrl"]

def buat_dlq(nama: str) -> str:
    """Create a Dead Letter Queue, return the ARN."""
    response = sqs.create_queue(
        QueueName=nama,
        Attributes={
            "MessageRetentionPeriod": str(14 * 24 * 3600)  # keep for 14 days
        }
    )
    url = response["QueueUrl"]
    # Get the ARN of the newly created queue
    attrs = sqs.get_queue_attributes(QueueUrl=url, AttributeNames=["QueueArn"])
    return attrs["Attributes"]["QueueArn"]

def ambil_queue_url(nama: str) -> str:
    """Get the URL of an existing queue."""
    response = sqs.get_queue_url(QueueName=nama)
    return response["QueueUrl"]

import json
from datetime import datetime, timezone
import uuid

QUEUE_URL = ambil_queue_url("order-queue")

def kirim_pesan(queue_url: str, payload: dict, delay_detik: int = 0) -> str:
    """Send one message to a Standard Queue."""
    body = json.dumps(payload, ensure_ascii=False)

    response = sqs.send_message(
        QueueUrl=queue_url,
        MessageBody=body,
        DelaySeconds=delay_detik,         # delay delivery (0–900 seconds)
        MessageAttributes={
            "source": {
                "DataType":    "String",
                "StringValue": "order-service"
            },
            "event_type": {
                "DataType":    "String",
                "StringValue": payload.get("event", "unknown")
            }
        }
    )
    msg_id = response["MessageId"]
    print(f"Message sent: {msg_id}")
    return msg_id

def kirim_pesan_fifo(queue_url: str, payload: dict, group_id: str) -> str:
    """Send a message to a FIFO Queue with a Message Group ID."""
    body = json.dumps(payload, ensure_ascii=False)

    response = sqs.send_message(
        QueueUrl=queue_url,
        MessageBody=body,
        MessageGroupId=group_id,               # messages in the same group are guaranteed ordered
        MessageDeduplicationId=str(uuid.uuid4())  # unique ID to prevent duplicates
    )
    return response["MessageId"]

# Usage example
order = {
    "event":    "order.created",
    "order_id": 1001,
    "user_id":  42,
    "total":    18500000,
    "timestamp": datetime.now(timezone.utc).isoformat()
}

kirim_pesan(QUEUE_URL, order)

# FIFO -- messages per user_id are guaranteed ordered
kirim_pesan_fifo(
    ambil_queue_url("order-fifo.fifo"),
    order,
    group_id=f"user-{order['user_id']}"  # all orders for user 42 processed in order
)

def kirim_batch(queue_url: str, payloads: list[dict]) -> dict:
    """
    Send up to 10 messages at once.
    More efficient than sending one by one (AWS billing is per request).
    """
    entries = [
        {
            "Id":          str(i),
            "MessageBody": json.dumps(p, ensure_ascii=False),
            "MessageAttributes": {
                "event_type": {
                    "DataType":    "String",
                    "StringValue": p.get("event", "unknown")
                }
            }
        }
        for i, p in enumerate(payloads[:10])   # max 10 per batch
    ]

    response = sqs.send_message_batch(QueueUrl=queue_url, Entries=entries)

    sukses = len(response.get("Successful", []))
    gagal  = len(response.get("Failed", []))

    if response.get("Failed"):
        for f in response["Failed"]:
            print(f"Failed to send ID {f['Id']}: {f['Message']}")

    print(f"Batch sent: {sukses} success, {gagal} failed")
    return response

# Send 10 orders at once
orders = [{"event": "order.created", "order_id": i, "total": i * 10000} for i in range(1, 11)]
kirim_batch(QUEUE_URL, orders)

import signal
import json
import time

def proses_order(payload: dict) -> None:
    """Business logic -- if an exception is raised, the message isn't deleted."""
    print(f"Processing order #{payload['order_id']} — Rp{payload['total']:,.0f}")
    # ... save to the database, send notifications, etc.

def jalankan_consumer(queue_url: str) -> None:
    berjalan = True

    def handle_shutdown(signum, frame):
        nonlocal berjalan
        print("Shutdown signal, stopping consumer...")
        berjalan = False

    signal.signal(signal.SIGINT,  handle_shutdown)
    signal.signal(signal.SIGTERM, handle_shutdown)

    print(f"Consumer active, polling from the queue...")

    while berjalan:
        try:
            # Long Polling: wait up to 20 seconds if the queue is empty
            # Far more efficient than rapid polling (reduces API call cost)
            response = sqs.receive_message(
                QueueUrl=queue_url,
                MaxNumberOfMessages=10,       # fetch up to 10 messages per request
                WaitTimeSeconds=20,           # Long Polling
                VisibilityTimeout=30,         # override the visibility timeout for this batch
                MessageAttributeNames=["All"],
                AttributeNames=["All"]
            )

            messages = response.get("Messages", [])
            if not messages:
                continue   # empty queue, poll again

            for msg in messages:
                receipt_handle = msg["ReceiptHandle"]
                msg_id         = msg["MessageId"]

                try:
                    payload = json.loads(msg["Body"])
                    proses_order(payload)

                    # REQUIRED: delete after successful processing
                    # If not deleted, the message reappears after the VisibilityTimeout
                    sqs.delete_message(
                        QueueUrl=queue_url,
                        ReceiptHandle=receipt_handle
                    )
                    print(f"✓ Message {msg_id} processed and deleted.")

                except json.JSONDecodeError as e:
                    print(f"✗ Invalid message format: {e}")
                    # Delete the invalid message -- retrying won't fix it
                    sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=receipt_handle)

                except Exception as e:
                    print(f"✗ Error processing {msg_id}: {e}")
                    # DON'T delete -- let the VisibilityTimeout expire
                    # The message will reappear to be processed again
                    # After maxReceiveCount failures, it goes to the DLQ

        except Exception as e:
            print(f"Error polling: {e}")
            time.sleep(5)   # pause before retrying on network errors

    print("Consumer stopped.")

jalankan_consumer(QUEUE_URL)

def consumer_batch_delete(queue_url: str) -> None:
    """
    Consumer with batch delete -- more efficient for high throughput.
    Process all messages, collect the successful ones, delete at once.
    """
    response = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=10,
        WaitTimeSeconds=20
    )
    messages = response.get("Messages", [])
    if not messages:
        return

    berhasil_dihapus = []

    for msg in messages:
        try:
            payload = json.loads(msg["Body"])
            proses_order(payload)
            berhasil_dihapus.append({
                "Id":            msg["MessageId"],
                "ReceiptHandle": msg["ReceiptHandle"]
            })
        except Exception as e:
            print(f"✗ Failed to process {msg['MessageId']}: {e}")
            # Not added to berhasil_dihapus -- will retry automatically

    # Batch delete for the successful messages
    if berhasil_dihapus:
        sqs.delete_message_batch(
            QueueUrl=queue_url,
            Entries=berhasil_dihapus
        )
        print(f"Batch delete: {len(berhasil_dihapus)} messages deleted.")

def proses_dlq(dlq_url: str, kirim_alert: bool = True) -> None:
    """
    Read messages from the Dead Letter Queue for monitoring and debugging.
    Messages in the DLQ are those that failed maxReceiveCount times.
    """
    response = sqs.receive_message(
        QueueUrl=dlq_url,
        MaxNumberOfMessages=10,
        WaitTimeSeconds=5,
        AttributeNames=["All"]
    )
    messages = response.get("Messages", [])

    if not messages:
        print("DLQ is empty.")
        return

    print(f"⚠ {len(messages)} messages in the DLQ:")
    for msg in messages:
        attrs        = msg.get("Attributes", {})
        receive_count = attrs.get("ApproximateReceiveCount", "?")

        try:
            payload = json.loads(msg["Body"])
        except Exception:
            payload = msg["Body"]

        print(f"  ID: {msg['MessageId']}")
        print(f"  Attempted: {receive_count}x")
        print(f"  Payload: {payload}")

        if kirim_alert:
            # Send a notification to the team (Slack, email, PagerDuty, etc.)
            print(f"  → Alert sent for message {msg['MessageId']}")

    # After analysis, you can:
    # 1. Delete messages from the DLQ (if unrecoverable)
    # 2. Move them back to the main queue after the bug is fixed (redrive)

def redrive_dari_dlq(dlq_url: str, main_queue_url: str) -> int:
    """Move messages from the DLQ back to the main queue after the bug is fixed."""
    dipindah = 0
    while True:
        response = sqs.receive_message(
            QueueUrl=dlq_url,
            MaxNumberOfMessages=10,
            WaitTimeSeconds=1
        )
        messages = response.get("Messages", [])
        if not messages:
            break

        for msg in messages:
            sqs.send_message(QueueUrl=main_queue_url, MessageBody=msg["Body"])
            sqs.delete_message(QueueUrl=dlq_url, ReceiptHandle=msg["ReceiptHandle"])
            dipindah += 1

    print(f"{dipindah} messages moved from the DLQ to the main queue.")
    return dipindah

def proses_dengan_extend_visibility(queue_url: str) -> None:
    """
    For messages that need long processing time,
    extend the visibility timeout periodically so they don't reappear.
    """
    response = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=1,
        WaitTimeSeconds=20,
        VisibilityTimeout=30   # give 30 initial seconds
    )
    messages = response.get("Messages", [])
    if not messages:
        return

    msg            = messages[0]
    receipt_handle = msg["ReceiptHandle"]

    try:
        payload = json.loads(msg["Body"])
        print(f"Processing a large message: {msg['MessageId']}")

        for langkah in range(5):
            # Simulate a long process
            time.sleep(20)

            # Extend the visibility timeout before it expires
            sqs.change_message_visibility(
                QueueUrl=queue_url,
                ReceiptHandle=receipt_handle,
                VisibilityTimeout=30   # add 30 more seconds
            )
            print(f"Visibility timeout extended, step {langkah + 1}/5")

        # Done -- delete the message
        sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=receipt_handle)
        print("Message processed successfully.")

    except Exception as e:
        print(f"Error: {e}")
        # Don't extend -- let the timeout expire and the message reappear