Boto3 SNS Tutorial

Amazon Simple Notification Service (Amazon SNS) is a highly available, secure, and fully managed Publisher-Subscriber (pub/sub) messaging service that allows application-to-application (A2A) and application-to-person (A2P) communication. Amazon SNS enables us to publish messages and distribute them to many subscriber systems; this includes HTTPS endpoints, webhooks, and AWS services such as AWS Lambda functions, Amazon SQS queues, and Amazon Kinesis Data Firehose. Additionally, services can use SNS to fan out notifications to end users using mobile push, SMS, and email. This Boto3 SNS tutorial covers how to work with Amazon SNS to create, list, delete topics, and send messages to them using Python.

Prerequisites

To start interacting with Amazon SNS programmatically and making API calls to manage SNS topics and subscriptions, you must first configure your Python environment.

In general, here’s what you need to have installed:

• Python 3
• Boto3
• AWS CLI tools

Alternatively, you can use a Cloud9 IDE.

What is Amazon SNS?

Amazon Simple Notification Service (Amazon SNS) is an AWS service that allows publishing (or sending) messages to multiple destinations (subscribers). In other words, the Amazon Simple Notification Service manages and coordinates the delivery of push messages to clients or subscribed endpoints.

Moreover, Amazon SNS supports endpoints and endpoint attributes for devices from the supported push notification service list. For example, GCM (Firebase Cloud Messaging) is a supported push notification service. The APNS (Apple Push Notification Service) is supported too.

Amazon SNS allows you to distribute messages, events, or notifications using your preferred programming language to the following subscriber types:

• Amazon Kinesis Firehorse
• Amazon SQS
• AWS Lambda
• Email
• HTTP/HTTPS
• Platform application endpoint
• SMS

Here’s an example of the typical fan-out architecture that distributes file upload events from the S3 bucket to multiple Lambda functions. Each of these can convert uploaded files to another format.

The application-to-application (A2A) pub/sub functionality provides topics for high-throughput, push-based, many-to-many messaging between distributed systems, microservices, and event-driven serverless applications.

As a result of using Amazon SNS topics, publisher systems can fan-out messages to many subscriber systems, including Amazon SQS queues, AWS Lambda functions, and HTTPS endpoints. The A2P functionality lets you publish SMS messages at scale via SMS, do mobile push notifications, and organize email notifications.

Benefits of using SNS

Here’s a list of benefits of using Amazon SNS:

• Scalability – Amazon Simple Notification Service topics scale up to any number of publishers, subscribers, and messages
• Ease of setup – Simple Notification Service is a fully managed service by AWS, therefore setting it up requires zero infrastructure work
• Multiple notification options – Amazon Simple Notification Service supports AWS Lambda, AWS SQS notifications, mobile push notifications, HTTP(S) endpoints, webhooks, email addresses, and SMS message.
• Integration with AWS Lambda – The native integration with AWS Lambda allows you to run a Lambda function every time a message is published to an SNS topic

How to connect to Amazon SNS using Boto3?

The Boto3 library provides you with two ways to access APIs for managing AWS services:

• The Boto3 client allows you to access the low-level API data. For example, you can access API response data in JSON format.
• The Boto3 resource allows you to use AWS services in a higher-level object-oriented way. For more information on the topic, look at AWS CLI vs. botocore vs. Boto3.

Here’s how we can instantiate the Boto3 SNS client to start working with Amazon SNS APIs:

import boto3
AWS_REGION = "us-east-1"
sns_client = boto3.client("sns", region_name=AWS_REGION)

Similarly, you can instantiate the Boto3 SNS resource:

import boto3
AWS_REGION = "us-east-1"
sns_resource = boto3.resource("sns", region_name=AWS_REGION)

Managing SNS topics using Boto3

The primary purpose of a publish/subscribe system is to allow message distribution from the application or service to many possible destinations. The source application is sending a message to a topic. Destination applications or services are subscribed to the topic and receive the same message every time the message is pushed to it.

Create SNS topic

To create an SNS topic, you need to use the create_topic() method of the Boto3 SNS client.

Amazon SNS topic is a communication channel that allows you to send the same message to multiple destinations (AWS Lambda functions, Amazon SQS queues, HTTPS endpoints, webhooks, and Amazon Kinesis Data Firehose) through a topic.

import logging
import boto3
from botocore.exceptions import ClientError
AWS_REGION = 'us-east-1'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
sns_client = boto3.client('sns', region_name=AWS_REGION)

def create_topic(name):
    """
    Creates a SNS notification topic.
    """
    try:
        topic = sns_client.create_topic(Name=name)
        logger.info(f'Created SNS topic {name}.')
    except ClientError:
        logger.exception(f'Could not create SNS topic {name}.')
        raise
    else:
        return topic

if __name__ == '__main__':
    topic_name = 'hands-on-cloud-sns-topic'
    logger.info(f'Creating SNS topic {topic_name}...')
    topic = create_topic(topic_name)

Thecreate_topic() method requires at least one argument:

• Name – the name of the topic. The topic name must contain only uppercase/lowercase ASCII letters, numbers, underscores, and hyphens and must be between 1 and 256 characters long.

We’re using the Python standard logging module to log messages to standard output in the above example.

Also, the botocore.exceptions module allows effective handle the Boto3 exceptions and errors in the script. We recommend you check our article about Exceptions Handling in Python for more information.

Here’s an execution output:

List SNS topics

To get the list of SNS topics, you need to use the list_topics() method from the Boto3 library.

We will use the Boto3 library paginator object to get the complete output from the list_topics() method.

Some AWS requests return incomplete output, therefore, require subsequent requests to get the complete result. The process of sending subsequent requests to continue where a previous request left off is called pagination.

import logging
import boto3
from botocore.exceptions import ClientError
AWS_REGION = 'us-east-1'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
sns_client = boto3.client('sns', region_name=AWS_REGION)

def list_topics():
    """
    Lists all SNS notification topics using paginator.
    """
    try:
        paginator = sns_client.get_paginator('list_topics')
        # creating a PageIterator from the paginator
        page_iterator = paginator.paginate().build_full_result()
        topics_list = []
        # loop through each page from page_iterator
        for page in page_iterator['Topics']:
            topics_list.append(page['TopicArn'])
    except ClientError:
        logger.exception(f'Could not list SNS topics.')
        raise
    else:
        return topics_list

if __name__ == '__main__':
    logger.info(f'Listing all SNS topics...')
    topics = list_topics()
    for topic in topics:
        logger.info(topic)

The list_topics() method does not require any argument as parameter input. It returns a list of SNS topics and the topic Amazon Resource Number (ARN).

Here’s an execution output:

List AWS SNS topic attributes

To get the AWS SNS topic attributes, you need to use the get_topic_attributes() method.

This method returns all of the properties of a topic:

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-1'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
sns_client = boto3.client('sns', region_name=AWS_REGION)

def list_topic_attributes():
    """
    Lists all SNS topics attributes using paginator.
    """
    try:
        paginator = sns_client.get_paginator('list_topics')
        # creating a PageIterator from the paginator
        page_iterator = paginator.paginate().build_full_result()
        topic_attributes_list = []
        # loop through each page from page_iterator
        for page in page_iterator['Topics']:
            
            response = sns_client.get_topic_attributes(
                TopicArn=page['TopicArn']
            )['Attributes']
            
            dict_obj = {
                'TopicArn': page['TopicArn'],
                'TopicPolicy': json.loads(response['Policy'])
            }
            
            topic_attributes_list.append(dict_obj)
            
    except ClientError:
        logger.exception(f'Could not get SNS topic attributes.')
        raise
    else:
        return topic_attributes_list

if __name__ == '__main__':
    logger.info(f'Listing all SNS topic attributes ...')
    topic_attributes = list_topic_attributes()
    for topic_attribute in topic_attributes:
        logger.info(json.dumps(topic_attribute, indent=4, sort_keys=True))

The required argument is:

• TopicArn – The ARN of the topic whose properties you want to get.

To get attributes of all the topics, we will use the list_topics() method paginator, which returns the ARN of all the topics. ARN returned by the list_topics() method is passed to get_topic_attributes().

In addition to that, we’re using the JSON module to parse the output from get_topic_attributes() and list_topics() methods.

Here’s an execution output:

Set SNS topic policy

To add a policy statement to a topic’s access control policy and grant access for specified identities to specific actions, you need to use the add_permission() method from the Boto3 library.

For more information we recommend you to check out the Working with IAM in Python using the Boto3 article.

import logging
import boto3
from botocore.exceptions import ClientError
AWS_REGION = 'us-east-1'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
sns_client = boto3.client('sns', region_name=AWS_REGION)

def add_permission(topic_arn, policy_label, account_ids, action_names):
    """
    Adds a policy statement to a topic's access control policy.
    """
    try:
        response = sns_client.add_permission(TopicArn=topic_arn,
                                             Label=policy_label,
                                             AWSAccountId=account_ids,
                                             ActionName=action_names)
    except ClientError:
        logger.exception(f'Could not add access policy to the topic.')
        raise
    else:
        return response

if __name__ == '__main__':
    topic_arn = 'your-topic-ARN'
    policy_label = 'hands-on-cloud-sns-topic-policy'
    account_ids = ['your-account-id']
    action_names = ['Publish', 'GetTopicAttributes']
    logger.info(f'Adding access policy {policy_label} to {topic_arn}...')
    add_permission(topic_arn, policy_label, account_ids, action_names)
    logger.info(f'Access policy {policy_label} added to {topic_arn}.')

The required arguments are:

• TopicArn – The topic ARN whose access control policy you need to modify.
• Label – The unique identifier for the new policy statement for the topic.
• AWSAccountId – The account IDs of the users (principals) who will be given access to the specified actions.
• ActionName – The action you want to allow for the specified principal(s).

This method does not return any response upon successfully modifying the access policy.

Here’s an execution output:

Publish message to SNS topic

To publish messages to the SNS topic, you need to use the publish() method from the Boto3 SNS client.

Let’s take a look at how to publish a message on an SNS topic that will get delivered to all subscribers:

import logging
import boto3
from botocore.exceptions import ClientError
AWS_REGION = 'us-east-1'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
sns_client = boto3.client('sns', region_name=AWS_REGION)

def publish_message(topic_arn, message, subject):
    """
    Publishes a message to a topic.
    """
    try:
        response = sns_client.publish(
            TopicArn=topic_arn,
            Message=message,
            Subject=subject,
        )['MessageId']
    except ClientError:
        logger.exception(f'Could not publish message to the topic.')
        raise
    else:
        return response

if __name__ == '__main__':
    topic_arn = 'your-topic-ARN'
    message = 'This is a test message on topic.'
    subject = 'This is a message subject on topic.'
    logger.info(f'Publishing message to topic - {topic_arn}...')
    message_id = publish_message(topic_arn, message, subject)
    logger.info(
        f'Message published to topic - {topic_arn} with message Id - {message_id}.'
    )

The required argument is:

• Message – The message you want to send.

The optional arguments we have used in the above example:

• TopicArn – the ARN of the SNS topic you want to publish the message.

• Subject  – an optional parameter to be used as the “Subject” line when the message is delivered to email endpoints.

The publish() method returns a message Id of the published message to the SNS topic.

Here’s an execution output:

Delete SNS topic

To delete an SNS topic, you need to use the delete_topic() method from the Boto3 library:

import logging
import boto3
from botocore.exceptions import ClientError
AWS_REGION = 'us-east-1'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
sns_client = boto3.client('sns', region_name=AWS_REGION)

def delete_topic(topic_arn):
    """
    Delete a SNS topic.
    """
    try:
        response = sns_client.delete_topic(TopicArn=topic_arn)
    except ClientError:
        logger.exception(f'Could not delete a SNS topic.')
        raise
    else:
        return response

if __name__ == '__main__':
    topic_arn = 'your-topic-ARN'
    logger.info(f'Deleting a SNS topic...')
    delete_response = delete_topic(topic_arn)
    logger.info(f'Deleted a topic - {topic_arn} successfully.')

The required argument is:

• TopicArn – The ARN of the SNS topic you want to delete.

The delete_topic() does not return any response after the successful deletion of the topic.

Here’s an execution output:

Managing Amazon SNS topic subscription

Before sending a message to consumers, we need to subscribe them to the topic (phone number, email, HTTP/S Endpoint, Lambda, or SQS).

Create an Amazon SNS subscription

To subscribe to an Amazon SNS web service topic programmatically, you need to use the subscribe() method of the Boto3 library:

import logging
import boto3
from botocore.exceptions import ClientError
AWS_REGION = 'us-east-1'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
sns_client = boto3.client('sns', region_name=AWS_REGION)

def subscribe(topic, protocol, endpoint):
    """
    Subscribe to a topic using endpoint as email OR SMS
    """
    try:
        subscription = sns_client.subscribe(
            TopicArn=topic,
            Protocol=protocol,
            Endpoint=endpoint,
            ReturnSubscriptionArn=True)['SubscriptionArn']
    except ClientError:
        logger.exception(
            "Couldn't subscribe {protocol} {endpoint} to topic {topic}.")
        raise
    else:
        return subscription

if __name__ == '__main__':
    topic_arn = 'your-topic-ARN'
    protocol = 'email'
    endpoint = 'your-email@mail.com'
    logger.info('Subscribing to a SNS topic...')
    # Creates an email subscription
    response = subscribe(topic_arn, protocol, endpoint)
    logger.info(
        f'Subscribed to a topic successfully.\nSubscription Arn - {response}')

The required arguments are:

• TopicArn: The topic ARN you want to subscribe to.

• Protocol: The endpoint protocol, such as ‘sms’ or ’email.’

An optional argument used in the above example is:

• Endpoint: The endpoint that receives messages, such as a phone number or an email address (for integration with the app and mobile device, for example).

The subscribe() returns Subscription Arn as a response after the successful subscription of the topic.

Here’s an execution output:

List SNS topic subscriptions

To get the list of all subscriptions on a topic, you need to use the list_subscriptions_by_topic() method of the Boto3 library.

By default, the list_subscriptions_by_topic() method returns a limited list of subscriptions, up to 100. To get the complete list of subscriptions on a topic, we will use the Boto3 paginator.

import logging
import boto3
from botocore.exceptions import ClientError
import json
AWS_REGION = 'us-east-1'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
sns_client = boto3.client('sns', region_name=AWS_REGION)

def list_topic_subscriptions(topic_arn):
    """
    Lists SNS notification topic subscriptions using paginator.
    """
    try:
        paginator = sns_client.get_paginator('list_subscriptions_by_topic')
        # creating a PageIterator from the paginator
        page_iterator = paginator.paginate(TopicArn=topic_arn,
                                           PaginationConfig={'MaxItems': 100})
        topic_subscriptions = []
        # loop through each page from page_iterator
        for page in page_iterator:
            for subscription in page['Subscriptions']:
                topic_subscriptions.append(subscription)
    except ClientError:
        logger.exception(f'Could not list SNS topics.')
        raise
    else:
        return topic_subscriptions

if __name__ == '__main__':
    topic_arn = 'your-topic-ARN'
    logger.info(f'Listing SNS topic subscriptions...')
    topic_subscriptions = list_topic_subscriptions(topic_arn)
    for subscription in topic_subscriptions:
        logger.info(json.dumps(subscription, indent=4, sort_keys=True))

The required argument is:

• TopicArn: The topic ARN you want to subscribe to.

The list_subscriptions_by_topic() returns the subscriptions dictionary object as a response after the successful subscription of the topic. 

SubscriptionArn, Owner, Protocol, Endpoint, and TopicArn of the topic can be parsed from this response object.

Here’s an execution output:

Delete SNS topic subscription

To unsubscribe or delete an SNS topic subscription, you need to use the unsubscribe() method from the Boto3 library.

If the subscription requires authentication for deletion, only the owner or the topic’s owner can use the unsubscribe() method.

The unsubscribe() method is throttled at 100 transactions per second (TPS) by default.

import logging
import boto3
from botocore.exceptions import ClientError
AWS_REGION = 'us-east-1'
# logger config
logger = logging.getLogger()
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s: %(levelname)s: %(message)s')
sns_client = boto3.client('sns', region_name=AWS_REGION)

def unsubscribe(subscription_arn):
    """
    Delete/ Unsubscribe to a topic subscription
    """
    try:
        unsubscribe = sns_client.unsubscribe(SubscriptionArn=subscription_arn)
    except ClientError:
        logger.exception("Couldn't unsubscribe to {subscription_arn}.")
        raise
    else:
        return unsubscribe

if __name__ == '__main__':
    subscription_arn = 'your-subscription-arn'
    logger.info('Unsubscribing to a SNS topic subscription...')
    response = unsubscribe(subscription_arn)
    logger.info(
        f'Successfully unsubscribed to a topic.\nSubscription Arn - {subscription_arn}'
    )

The required argument is:

• SubscriptionArn: The subscription ARN of the topic to be deleted.

The unsubscribe() method does not return an output response after the successful unsubscription of the topic. 

Here’s an execution output:

Amazon SNS service limits

Amazon SNS has a few limits you should know when planning a system that uses SNS in the production environment.

For example, the email delivery rate has a max limit of 10 messages per second.

The subscribe and unsubscribe calls are limited to 100 transactions per second (per AWS account).

See the AWS documentation on SNS service limits for more details.

Summary

In conclusion, in this article, we’ve coveredhow to use Python and Boto3 library to interact with Amazon SNS to create, describe, list, and delete SNS topics and send messages to them.