rboto-sqs

rboto is an independent project maintained by Ferrumio. It is not affiliated with, sponsored by, or endorsed by Amazon Web Services (AWS).

Preview (Alpha). rboto is under active development. APIs may change between releases.

rboto provides an async-only, typed Python client for Amazon SQS, powered by Rust. This package contains SQSClient and 23 operations.

Installation

Requires Python 3.12 or newer and a native wheel for your Python version and platform.

python -m pip install "rboto[sqs]"

You can also install rboto-sqs directly; it installs the matching rboto package. Only the selected service is installed.

These are the package names to use when a release is available on your package index. If your index has no release yet, follow the source installation instructions.

Usage

Configure credentials through your environment, shared profile, or workload role. The example uses us-east-1; select a region where your service and resources are available. Resource names, identifiers, input files, and model IDs in the example are placeholders to replace with your own.

import asyncio

from rboto import SQSClient


async def main() -> None:
    client = SQSClient(region="us-east-1")
    response = await client.send_message(
        queue_url="https://sqs.us-east-1.amazonaws.com/123456789012/orders",
        message_body='{"order_id":"123"}',
    )
    print(response.message_id)


if __name__ == "__main__":
    asyncio.run(main())

In an application with an event loop, call await main() instead of asyncio.run(main()). Reuse the client across calls. Constructing a client does not send a request.

API conventions

• Import the client with from rboto import SQSClient.

• Every operation is asynchronous: use await client.method(...).

• Operation arguments are keyword-only and use the Python names listed below.

• Structured inputs use typed dictionaries and tagged unions from rboto_sqs.types.

• Regular responses expose typed attributes; response.to_dict() converts them to a dictionary.

• Event-stream responses may be typed dictionaries containing async iterators. Follow the declared return type; consume event streams with async for.

• Byte streams are consumed asynchronously. await stream.read() buffers the complete body; iterate the stream when processing a large response.

• Direct operation calls return one page.

• Use async for page in client.paginators.method(...) for supported paginated operations. Paginators accept the same typed arguments as the corresponding client method.

• Catch modeled exceptions from rboto_sqs.exceptions. Their error_code, message, and operation_name identify service failures.

Clients also accept profile, endpoint_url, or shared Session settings. See configuration, errors, and streaming.

Operations

This is the complete operation catalog for this generated package. “Required arguments” lists arguments without defaults in the Python signature. Service-side conditions may require additional arguments. Each method links to its complete signature, including optional arguments. Nested input definitions are in rboto_sqs.types; native response definitions are in the package's _native.pyi.

Method	Required arguments	Return type
add_permission	queue_url: str, label: str, aws_account_ids: Sequence[str], actions: Sequence[str]	None
cancel_message_move_task	task_handle: str	CancelMessageMoveTaskOutput
change_message_visibility	queue_url: str, receipt_handle: str, visibility_timeout: int	None
change_message_visibility_batch	queue_url: str, entries: Sequence[ChangeMessageVisibilityBatchRequestEntry]	ChangeMessageVisibilityBatchOutput
create_queue	queue_name: str	CreateQueueOutput
delete_message	queue_url: str, receipt_handle: str	None
delete_message_batch	queue_url: str, entries: Sequence[DeleteMessageBatchRequestEntry]	DeleteMessageBatchOutput
delete_queue	queue_url: str	None
get_queue_attributes	queue_url: str	GetQueueAttributesOutput
get_queue_url	queue_name: str	GetQueueUrlOutput
list_dead_letter_source_queues	queue_url: str	ListDeadLetterSourceQueuesOutput
list_message_move_tasks	source_arn: str	ListMessageMoveTasksOutput
list_queue_tags	queue_url: str	ListQueueTagsOutput
list_queues	None	ListQueuesOutput
purge_queue	queue_url: str	None
receive_message	queue_url: str	ReceiveMessageOutput
remove_permission	queue_url: str, label: str	None
send_message	queue_url: str, message_body: str	SendMessageOutput
send_message_batch	queue_url: str, entries: Sequence[SendMessageBatchRequestEntry]	SendMessageBatchOutput
set_queue_attributes	queue_url: str, attributes: dict[QueueAttributeName, str]	None
start_message_move_task	source_arn: str	StartMessageMoveTaskOutput
tag_queue	queue_url: str, tags: dict[str, str]	None
untag_queue	queue_url: str, tag_keys: Sequence[str]	None

Complete signatures

add_permission

async def add_permission(
    *,
    queue_url: str,
    label: str,
    aws_account_ids: Sequence[str],
    actions: Sequence[str],
) -> None: ...

cancel_message_move_task

async def cancel_message_move_task(
    *,
    task_handle: str,
) -> CancelMessageMoveTaskOutput: ...

change_message_visibility

async def change_message_visibility(
    *,
    queue_url: str,
    receipt_handle: str,
    visibility_timeout: int,
) -> None: ...

change_message_visibility_batch

async def change_message_visibility_batch(
    *,
    queue_url: str,
    entries: Sequence[ChangeMessageVisibilityBatchRequestEntry],
) -> ChangeMessageVisibilityBatchOutput: ...

create_queue

async def create_queue(
    *,
    queue_name: str,
    attributes: dict[QueueAttributeName, str] | None = None,
    tags: dict[str, str] | None = None,
) -> CreateQueueOutput: ...

delete_message

async def delete_message(
    *,
    queue_url: str,
    receipt_handle: str,
) -> None: ...

delete_message_batch

async def delete_message_batch(
    *,
    queue_url: str,
    entries: Sequence[DeleteMessageBatchRequestEntry],
) -> DeleteMessageBatchOutput: ...

delete_queue

async def delete_queue(
    *,
    queue_url: str,
) -> None: ...

get_queue_attributes

async def get_queue_attributes(
    *,
    queue_url: str,
    attribute_names: Sequence[QueueAttributeName] | None = None,
) -> GetQueueAttributesOutput: ...

get_queue_url

async def get_queue_url(
    *,
    queue_name: str,
    queue_owner_aws_account_id: str | None = None,
) -> GetQueueUrlOutput: ...

list_dead_letter_source_queues

async def list_dead_letter_source_queues(
    *,
    queue_url: str,
    next_token: str | None = None,
    max_results: int | None = None,
) -> ListDeadLetterSourceQueuesOutput: ...

Also available as client.paginators.list_dead_letter_source_queues(...), returning an async iterator of the response type above.

list_message_move_tasks

async def list_message_move_tasks(
    *,
    source_arn: str,
    max_results: int | None = None,
) -> ListMessageMoveTasksOutput: ...

list_queue_tags

async def list_queue_tags(
    *,
    queue_url: str,
) -> ListQueueTagsOutput: ...

list_queues

async def list_queues(
    *,
    queue_name_prefix: str | None = None,
    next_token: str | None = None,
    max_results: int | None = None,
) -> ListQueuesOutput: ...

Also available as client.paginators.list_queues(...), returning an async iterator of the response type above.

purge_queue

async def purge_queue(
    *,
    queue_url: str,
) -> None: ...

receive_message

async def receive_message(
    *,
    queue_url: str,
    attribute_names: Sequence[QueueAttributeName] | None = None,
    message_system_attribute_names: Sequence[MessageSystemAttributeName] | None = None,
    message_attribute_names: Sequence[str] | None = None,
    max_number_of_messages: int | None = None,
    visibility_timeout: int | None = None,
    wait_time_seconds: int | None = None,
    receive_request_attempt_id: str | None = None,
) -> ReceiveMessageOutput: ...

remove_permission

async def remove_permission(
    *,
    queue_url: str,
    label: str,
) -> None: ...

send_message

async def send_message(
    *,
    queue_url: str,
    message_body: str,
    delay_seconds: int | None = None,
    message_attributes: dict[str, MessageAttributeValue] | None = None,
    message_system_attributes: dict[MessageSystemAttributeNameForSends, MessageSystemAttributeValue] | None = None,
    message_deduplication_id: str | None = None,
    message_group_id: str | None = None,
) -> SendMessageOutput: ...

send_message_batch

async def send_message_batch(
    *,
    queue_url: str,
    entries: Sequence[SendMessageBatchRequestEntry],
) -> SendMessageBatchOutput: ...

set_queue_attributes

async def set_queue_attributes(
    *,
    queue_url: str,
    attributes: dict[QueueAttributeName, str],
) -> None: ...

start_message_move_task

async def start_message_move_task(
    *,
    source_arn: str,
    destination_arn: str | None = None,
    max_number_of_messages_per_second: int | None = None,
) -> StartMessageMoveTaskOutput: ...

tag_queue

async def tag_queue(
    *,
    queue_url: str,
    tags: dict[str, str],
) -> None: ...

untag_queue

async def untag_queue(
    *,
    queue_url: str,
    tag_keys: Sequence[str],
) -> None: ...

Links

• Service guide
• Repository
• Issues
• Releases

Licensed under Apache-2.0.