Managing Amazon SNS topic subscription using Boto3

<–Working with Amazon SNS in Python using Boto3 SDK

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

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-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 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 = 'arn:aws:sns:us-east-2:770877723960:sns-topic-2'
    protocol = 'email'
    endpoint = 'jayarohith2000@gmail.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:

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

    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:

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-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 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 = 'arn:aws:sns:us-east-2:770877723960:sns-topic-2:29e964e9-034a-4b2a-8ef0-d12c22b0970d'

    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:

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

working with AWS SNS subscriptions Download

Create an Amazon SNS subscription

List SNS topic subscriptions

Delete SNS topic subscription