rboto-sns

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 SNS, powered by Rust. This package contains SNSClient and 42 operations.

Installation

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

python -m pip install "rboto[sns]"

You can also install rboto-sns 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 SNSClient


async def main() -> None:
    client = SNSClient(region="us-east-1")
    response = await client.publish(
        topic_arn="arn:aws:sns:us-east-1:123456789012:notifications",
        message="Your report is ready",
    )
    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 SNSClient.

• 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_sns.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_sns.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_sns.types; native response definitions are in the package's _native.pyi.

Method	Required arguments	Return type
add_permission	topic_arn: str, label: str, aws_account_id: Sequence[str], action_name: Sequence[str]	None
check_if_phone_number_is_opted_out	phone_number: str	CheckIfPhoneNumberIsOptedOutOutput
confirm_subscription	topic_arn: str, token: str	ConfirmSubscriptionOutput
create_platform_application	name: str, platform: str, attributes: dict[str, str]	CreatePlatformApplicationOutput
create_platform_endpoint	platform_application_arn: str, token: str	CreatePlatformEndpointOutput
create_sms_sandbox_phone_number	phone_number: str	CreateSMSSandboxPhoneNumberOutput
create_topic	name: str	CreateTopicOutput
delete_endpoint	endpoint_arn: str	None
delete_platform_application	platform_application_arn: str	None
delete_sms_sandbox_phone_number	phone_number: str	DeleteSMSSandboxPhoneNumberOutput
delete_topic	topic_arn: str	None
get_data_protection_policy	resource_arn: str	GetDataProtectionPolicyOutput
get_endpoint_attributes	endpoint_arn: str	GetEndpointAttributesOutput
get_platform_application_attributes	platform_application_arn: str	GetPlatformApplicationAttributesOutput
get_sms_attributes	None	GetSMSAttributesOutput
get_sms_sandbox_account_status	None	GetSMSSandboxAccountStatusOutput
get_subscription_attributes	subscription_arn: str	GetSubscriptionAttributesOutput
get_topic_attributes	topic_arn: str	GetTopicAttributesOutput
list_endpoints_by_platform_application	platform_application_arn: str	ListEndpointsByPlatformApplicationOutput
list_origination_numbers	None	ListOriginationNumbersOutput
list_phone_numbers_opted_out	None	ListPhoneNumbersOptedOutOutput
list_platform_applications	None	ListPlatformApplicationsOutput
list_sms_sandbox_phone_numbers	None	ListSMSSandboxPhoneNumbersOutput
list_subscriptions	None	ListSubscriptionsOutput
list_subscriptions_by_topic	topic_arn: str	ListSubscriptionsByTopicOutput
list_tags_for_resource	resource_arn: str	ListTagsForResourceOutput
list_topics	None	ListTopicsOutput
opt_in_phone_number	phone_number: str	OptInPhoneNumberOutput
publish	message: str	PublishOutput
publish_batch	topic_arn: str, publish_batch_request_entries: Sequence[PublishBatchRequestEntry]	PublishBatchOutput
put_data_protection_policy	resource_arn: str, data_protection_policy: str	None
remove_permission	topic_arn: str, label: str	None
set_endpoint_attributes	endpoint_arn: str, attributes: dict[str, str]	None
set_platform_application_attributes	platform_application_arn: str, attributes: dict[str, str]	None
set_sms_attributes	attributes: dict[str, str]	SetSMSAttributesOutput
set_subscription_attributes	subscription_arn: str, attribute_name: str	None
set_topic_attributes	topic_arn: str, attribute_name: str	None
subscribe	topic_arn: str, protocol: str	SubscribeOutput
tag_resource	resource_arn: str, tags: Sequence[Tag]	TagResourceOutput
unsubscribe	subscription_arn: str	None
untag_resource	resource_arn: str, tag_keys: Sequence[str]	UntagResourceOutput
verify_sms_sandbox_phone_number	phone_number: str, one_time_password: str	VerifySMSSandboxPhoneNumberOutput

Complete signatures

add_permission

async def add_permission(
    *,
    topic_arn: str,
    label: str,
    aws_account_id: Sequence[str],
    action_name: Sequence[str],
) -> None: ...

check_if_phone_number_is_opted_out

async def check_if_phone_number_is_opted_out(
    *,
    phone_number: str,
) -> CheckIfPhoneNumberIsOptedOutOutput: ...

confirm_subscription

async def confirm_subscription(
    *,
    topic_arn: str,
    token: str,
    authenticate_on_unsubscribe: str | None = None,
) -> ConfirmSubscriptionOutput: ...

create_platform_application

async def create_platform_application(
    *,
    name: str,
    platform: str,
    attributes: dict[str, str],
) -> CreatePlatformApplicationOutput: ...

create_platform_endpoint

async def create_platform_endpoint(
    *,
    platform_application_arn: str,
    token: str,
    custom_user_data: str | None = None,
    attributes: dict[str, str] | None = None,
) -> CreatePlatformEndpointOutput: ...

create_sms_sandbox_phone_number

async def create_sms_sandbox_phone_number(
    *,
    phone_number: str,
    language_code: LanguageCodeString | None = None,
) -> CreateSMSSandboxPhoneNumberOutput: ...

create_topic

async def create_topic(
    *,
    name: str,
    attributes: dict[str, str] | None = None,
    tags: Sequence[Tag] | None = None,
    data_protection_policy: str | None = None,
) -> CreateTopicOutput: ...

delete_endpoint

async def delete_endpoint(
    *,
    endpoint_arn: str,
) -> None: ...

delete_platform_application

async def delete_platform_application(
    *,
    platform_application_arn: str,
) -> None: ...

delete_sms_sandbox_phone_number

async def delete_sms_sandbox_phone_number(
    *,
    phone_number: str,
) -> DeleteSMSSandboxPhoneNumberOutput: ...

delete_topic

async def delete_topic(
    *,
    topic_arn: str,
) -> None: ...

get_data_protection_policy

async def get_data_protection_policy(
    *,
    resource_arn: str,
) -> GetDataProtectionPolicyOutput: ...

get_endpoint_attributes

async def get_endpoint_attributes(
    *,
    endpoint_arn: str,
) -> GetEndpointAttributesOutput: ...

get_platform_application_attributes

async def get_platform_application_attributes(
    *,
    platform_application_arn: str,
) -> GetPlatformApplicationAttributesOutput: ...

get_sms_attributes

async def get_sms_attributes(
    *,
    attributes: Sequence[str] | None = None,
) -> GetSMSAttributesOutput: ...

get_sms_sandbox_account_status

async def get_sms_sandbox_account_status() -> GetSMSSandboxAccountStatusOutput: ...

get_subscription_attributes

async def get_subscription_attributes(
    *,
    subscription_arn: str,
) -> GetSubscriptionAttributesOutput: ...

get_topic_attributes

async def get_topic_attributes(
    *,
    topic_arn: str,
) -> GetTopicAttributesOutput: ...

list_endpoints_by_platform_application

async def list_endpoints_by_platform_application(
    *,
    platform_application_arn: str,
    next_token: str | None = None,
) -> ListEndpointsByPlatformApplicationOutput: ...

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

list_origination_numbers

async def list_origination_numbers(
    *,
    next_token: str | None = None,
    max_results: int | None = None,
) -> ListOriginationNumbersOutput: ...

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

list_phone_numbers_opted_out

async def list_phone_numbers_opted_out(
    *,
    next_token: str | None = None,
) -> ListPhoneNumbersOptedOutOutput: ...

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

list_platform_applications

async def list_platform_applications(
    *,
    next_token: str | None = None,
) -> ListPlatformApplicationsOutput: ...

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

list_sms_sandbox_phone_numbers

async def list_sms_sandbox_phone_numbers(
    *,
    next_token: str | None = None,
    max_results: int | None = None,
) -> ListSMSSandboxPhoneNumbersOutput: ...

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

list_subscriptions

async def list_subscriptions(
    *,
    next_token: str | None = None,
) -> ListSubscriptionsOutput: ...

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

list_subscriptions_by_topic

async def list_subscriptions_by_topic(
    *,
    topic_arn: str,
    next_token: str | None = None,
) -> ListSubscriptionsByTopicOutput: ...

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

list_tags_for_resource

async def list_tags_for_resource(
    *,
    resource_arn: str,
) -> ListTagsForResourceOutput: ...

list_topics

async def list_topics(
    *,
    next_token: str | None = None,
) -> ListTopicsOutput: ...

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

opt_in_phone_number

async def opt_in_phone_number(
    *,
    phone_number: str,
) -> OptInPhoneNumberOutput: ...

publish

async def publish(
    *,
    message: str,
    topic_arn: str | None = None,
    target_arn: str | None = None,
    phone_number: str | None = None,
    subject: str | None = None,
    message_structure: str | None = None,
    message_attributes: dict[str, MessageAttributeValue] | None = None,
    message_deduplication_id: str | None = None,
    message_group_id: str | None = None,
) -> PublishOutput: ...

publish_batch

async def publish_batch(
    *,
    topic_arn: str,
    publish_batch_request_entries: Sequence[PublishBatchRequestEntry],
) -> PublishBatchOutput: ...

put_data_protection_policy

async def put_data_protection_policy(
    *,
    resource_arn: str,
    data_protection_policy: str,
) -> None: ...

remove_permission

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

set_endpoint_attributes

async def set_endpoint_attributes(
    *,
    endpoint_arn: str,
    attributes: dict[str, str],
) -> None: ...

set_platform_application_attributes

async def set_platform_application_attributes(
    *,
    platform_application_arn: str,
    attributes: dict[str, str],
) -> None: ...

set_sms_attributes

async def set_sms_attributes(
    *,
    attributes: dict[str, str],
) -> SetSMSAttributesOutput: ...

set_subscription_attributes

async def set_subscription_attributes(
    *,
    subscription_arn: str,
    attribute_name: str,
    attribute_value: str | None = None,
) -> None: ...

set_topic_attributes

async def set_topic_attributes(
    *,
    topic_arn: str,
    attribute_name: str,
    attribute_value: str | None = None,
) -> None: ...

subscribe

async def subscribe(
    *,
    topic_arn: str,
    protocol: str,
    endpoint: str | None = None,
    attributes: dict[str, str] | None = None,
    return_subscription_arn: bool | None = None,
) -> SubscribeOutput: ...

tag_resource

async def tag_resource(
    *,
    resource_arn: str,
    tags: Sequence[Tag],
) -> TagResourceOutput: ...

unsubscribe

async def unsubscribe(
    *,
    subscription_arn: str,
) -> None: ...

untag_resource

async def untag_resource(
    *,
    resource_arn: str,
    tag_keys: Sequence[str],
) -> UntagResourceOutput: ...

verify_sms_sandbox_phone_number

async def verify_sms_sandbox_phone_number(
    *,
    phone_number: str,
    one_time_password: str,
) -> VerifySMSSandboxPhoneNumberOutput: ...

Links

• Service guide
• Repository
• Issues
• Releases

Licensed under Apache-2.0.