Working with SNS in Python using Boto3

Abhinav Dumbre

Abhinav Dumbre

0
(0)

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 distribute messages 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 article will cover how to use Python and Boto3 library to interact with Amazon SNS to create, describe, list, and delete SNS topics and send messages to them.

Prerequisites

To start interacting with Amazon SNS programmatically and making API calls to manage SNS topics and manage its 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 SNS is an AWS service that allows sending messages to multiple destinations (subscribers). In other words, the AWS SNS service manages and coordinates the delivery of push messages to clients or subscribed endpoints.

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 which can convert uploaded files to another format.

Serverless Fan-Out Architecture - SNS

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 send messages/events to users at scale via SMS, mobile push notifications, and email.

Benefits of using SNS

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

  • Scalability – Amazon SNS topics scale up to any number of publishers, subscribers, and messages
  • Ease of setup – SNS is a fully managed service by AWS, therefore setting it up requires zero infrastructure work
  • Multiple notification options – SNS supports AWS Lambda, AWS SQS notifications, mobile push notifications, HTTP(S) endpoints, webhooks, email addresses, and SMS messages
  • 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 get access to 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, take a 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 SNS client of the Boto3 library.

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)

The create_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 effectively handle the Boto3 exceptions and errors in the script. We recommend you to check out our article about Exceptions Handling in Python for more information.

Here’s an execution output:

1. Create SNS Topic

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:

2a. List All Topics

List SNS topic attributes

To get the 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:

2b. List SNS topics attributes

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 successful modification of the access policy.

Here’s an execution output:

3. Set SNS Topic Policy

Publish message on SNS topic

To post a message on a topic, you need to use the publish() method from the Boto3 library.

Let’s take a look at how to publish a message on a 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 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 on a topic.

Here’s an execution output:

4. Publish a SNS Message

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 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:

5. Delete a SNS Topic

Managing SNS topic subscriptions

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

Create SNS topic subscription

To subscribe to an SNS 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 = '[email protected]'

    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 protocol of the endpoint, 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.

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

Here’s an execution output:

6. Subscribe to a Topic

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. 

SubscriptionArnOwnerProtocolEndpoint, and TopicArn of the topic can be parsed from this response object.

Here’s an execution output:

List a Topic Subscriptions

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:

8. Unsubscribe to a Topic

Amazon SNS service limits

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

For example, The delivery rate for emails has a max limit of 10 messages per second.

Both 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 covered how to use Python and Boto3 library to interact with Amazon SNS to create, describe, list, and delete SNS topics and send messages to them.

How useful was this post?

Click on a star to rate it!

As you found this post useful...

Follow us on social media!

We are sorry that this post was not useful for you!

Let us improve this post!

Tell us how we can improve this post?

Top rated Udemy Courses to improve you career

Subscribe to our updates

Like this article?

Share on facebook
Share on Facebook
Share on twitter
Share on Twitter
Share on linkedin
Share on Linkdin
Share on pinterest
Share on Pinterest

Want to be an author of another post?

We’re looking for skilled technical authors for our blog!

Leave a comment

If you’d like to ask a question about the code or piece of configuration, feel free to use https://codeshare.io/ or a similar tool as Facebook comments are breaking code formatting.