Managing Amazon SNS topics using Boto3

<–Working with Amazon SNS in Python using Boto3 SDK

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.

Creating 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-2'

# 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 = 'sns-topic-2'
    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 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:

Listing 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.

import logging
import boto3
from botocore.exceptions import ClientError

AWS_REGION = 'us-east-2'

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

Deleting 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-2'

# 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 = 'arn:aws:sns:us-east-2:770877723960:sns-topic-2'

    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:

You can find the managing amazon simple notification service topics codes in the below