SNS and SQS

Use SNSClient() to publish notifications and SQSClient() to send, receive, and delete queue messages. Install their service packages from the checkout.

Publish to an existing topic

from rboto import SNSClient


async def publish(client: SNSClient, topic_arn: str, message: str) -> str | None:
    response = await client.publish(topic_arn=topic_arn, message=message)
    return response.message_id

Configure the client for the topic's region. A message ID confirms acceptance by SNS; subscriber processing happens separately.

Send to a queue

from rboto import SQSClient
from rboto_sqs.types import MessageAttributeValue


async def send_message(client: SQSClient, queue_url: str) -> str | None:
    source: MessageAttributeValue = {"data_type": "String", "string_value": "orders"}
    response = await client.send_message(
        queue_url=queue_url,
        message_body="Order received",
        message_attributes={"source": source},
    )
    return response.message_id

The example uses a standard queue URL. FIFO queues have additional grouping and deduplication requirements; supply their operation parameters for your queue.

Receive and acknowledge

from rboto import SQSClient


async def receive(client: SQSClient, queue_url: str) -> None:
    response = await client.receive_message(
        queue_url=queue_url,
        max_number_of_messages=10,
        wait_time_seconds=10,
    )
    for message in response.messages or []:
        print(message.body)  # Replace with your application's processing.
        if message.receipt_handle is not None:
            await client.delete_message(queue_url=queue_url, receipt_handle=message.receipt_handle)

Replace print() with your processing and delete each message after that work succeeds. Receipt handles come from the receive response. Account for redelivery. Long polling waits up to wait_time_seconds; an empty response is normal. Batch methods report per-entry failures that need inspection before retrying.