This article is the second part of a series about AWS SNS. Be sure to read through all the posts by clicking on the links below:
AWS SNS: The Complete Guide – Part 1: Theory
AWS SNS: The Complete Guide – Part 2: AWS CLI SNS Commands
AWS SNS: The Complete Guide – Part 3: AWS SNS, Python and Boto3 Interface
AWS SNS: The Complete Guide – Part 4: AWS SNS Unit Testing Using Moto
AWS SNS: The Complete Guide – Part 5: AWS SNS using Terraform
AWS SNS: The Complete Guide – Part 6: AWS SNS FAQ

We discussed AWS SNS in theory and AWS-CLI-related concepts in the previous article. This article uses Boto3 to perform various AWS SNS operations like creating a topic, publishing data, etc.

How to create SNS Topic using Boto3?

from typing import List

import boto3

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


def create_sns_topic(name: str) -> dict:
    """Creates SNS topic.

    :param name: Name of SNS Topic to be created.
    :return: JSON containing TopicARN.
    """
    response = sns_client.create_topic(Name=name)
    return response

Sample Call
create_sns_topic(name="test_topic")

Sample Response
{'TopicArn': 'arn:aws:sns:us-east-1:123456789012:test_topic', 'ResponseMetadata': {'RequestId': '3e381117-6010-469f-a938-106948389a45', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': '3e381117-6010-469f-a938-106948389a45', 'content-type': 'text/xml', 'content-length': '317', 'date': 'Fri, 21 Jan 2022 10:59:09 GMT'}, 'RetryAttempts': 0}}

How to list all SNS Topics using Boto3?

from typing import List

import boto3

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


def list_sns_topics():
    response = sns_client.list_topics()
    return response

Sample Call
list_sns_topics()

Sample Response
{'Topics': [{'TopicArn': 'arn:aws:sns:us-east-1:123456789012:test_topic'}], 'ResponseMetadata': {'RequestId': '1cf1b070-4b34-4fe8-aa04-40305ca0710f', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': '1cf1b070-4b34-4fe8-aa04-40305ca0710f', 'content-type': 'text/xml', 'content-length': '375', 'date': 'Fri, 21 Jan 2022 10:59:09 GMT'}, 'RetryAttempts': 0}}

How to add permissions to existing SNS Topics using Boto3?

from typing import List

import boto3

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


def add_sns_permission(topic_arn: str, label: str, action: str, principal: str):
    """Adds permission to SNS Topic.

    :param topic_arn: ARN of SNS Topic to be added permission to.
    :param label: Label of SNS Topic to be added permission to.
    :param action: Action to be added to SNS Topic.
    :param principal: Principal to be added to SNS Topic.
    :return: JSON containing TopicARN.
    """
    response = sns_client.add_permission(
        TopicArn=topic_arn,
        Label=label,
        Action=action,
        Principal=principal,
    )
    return response

How to remove SNS Topic permissions using Boto3?

from typing import List

import boto3

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


def remove_sns_permission(topic_arn: str, label: str):
    """Removes permission from SNS Topic.

    :param topic_arn: ARN of SNS Topic to be removed permission from.
    :param label: Label of SNS Topic to be removed permission from.
    :return: JSON containing TopicARN.
    """
    response = sns_client.remove_permission(
        TopicArn=topic_arn,
        Label=label,
    )
    return response

How to set SNS Topic attributes using Boto3?

from typing import List

import boto3

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


def set_sns_topic_attributes(topic_arn: str, attribute_name: str, attribute_value: str):
    """Sets SNS Topic Attributes.

    :param topic_arn: ARN of SNS Topic to be set attributes.
    :param attribute_name: Name of attribute to be set.
    :param attribute_value: Value of attribute to be set.
    :return: JSON containing TopicARN.
    """
    response = sns_client.set_topic_attributes(
        TopicArn=topic_arn,
        AttributeName=attribute_name,
        AttributeValue=attribute_value,
    )
    return response

Sample Call
set_sns_topic_attributes("arn:aws:sns:us-east-1:123456789012:test_topic", attribute_name="DisplayName", attribute_value="test_display_name")

Sample Response
{'ResponseMetadata': {'RequestId': 'eeb218a7-194f-4556-bd98-2918bcee241b', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': 'eeb218a7-194f-4556-bd98-2918bcee241b', 'content-type': 'text/xml', 'content-length': '215', 'date': 'Fri, 21 Jan 2022 10:59:09 GMT'}, 'RetryAttempts': 0}}

How to get SNS Topic attributes using Boto3?

from typing import List

import boto3

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


def get_sns_topic_attributes(topic_arn: str):
    """Gets SNS Topic Attributes.

    :param topic_arn: ARN of SNS Topic to be get attributes.
    :return: JSON containing TopicARN.
    """
    response = sns_client.get_topic_attributes(
        TopicArn=topic_arn,
    )
    return response

Sample Call
get_sns_topic_attributes("arn:aws:sns:us-east-1:123456789012:test_topic")


Sample Response
{'Attributes': {'Policy': '{"Version":"2008-10-17","Id":"__default_policy_ID","Statement":[{"Sid":"__default_statement_ID","Effect":"Allow","Principal":{"AWS":"*"},"Action":["SNS:GetTopicAttributes","SNS:SetTopicAttributes","SNS:AddPermission","SNS:RemovePermission","SNS:DeleteTopic","SNS:Subscribe","SNS:ListSubscriptionsByTopic","SNS:Publish"],"Resource":"arn:aws:sns:us-east-1:123456789012:test_topic","Condition":{"StringEquals":{"AWS:SourceOwner":"123456789012"}}}]}', 'Owner': '123456789012', 'SubscriptionsPending': '0', 'TopicArn': 'arn:aws:sns:us-east-1:123456789012:test_topic', 'EffectiveDeliveryPolicy': '{"http":{"defaultHealthyRetryPolicy":{"minDelayTarget":20,"maxDelayTarget":20,"numRetries":3,"numMaxDelayRetries":0,"numNoDelayRetries":0,"numMinDelayRetries":0,"backoffFunction":"linear"},"disableSubscriptionOverrides":false}}', 'SubscriptionsConfirmed': '0', 'DisplayName': 'test_display_name', 'SubscriptionsDeleted': '0'}, 'ResponseMetadata': {'RequestId': '2bc61ab7-c791-4d0d-be17-acc43402cc63', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': '2bc61ab7-c791-4d0d-be17-acc43402cc63', 'content-type': 'text/xml', 'content-length': '2137', 'date': 'Fri, 21 Jan 2022 10:59:10 GMT'}, 'RetryAttempts': 0}}

How to add tags to SNS Topic using Boto3?

from typing import List

import boto3

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


def tag_sns_resource(topic_arn: str, tags: List[dict]):
    """Tags SNS Topic.

    :param topic_arn: ARN of SNS Topic to be tagged.
    :param tags: Dictionary of tags to be added to SNS Topic.
    :return: JSON containing TopicARN.
    """
    response = sns_client.tag_resource(
        ResourceArn=topic_arn,
        Tags=tags,
    )
    return response

Sample Call
tag_sns_resource("arn:aws:sns:us-east-1:123456789012:test_topic", tags=[{"Key": "test_tag_key", "Value": "test_tag_value"}])


Sample Response
{'ResponseMetadata': {'RequestId': 'bcaff823-c5ca-4700-8c2e-811b4c792da9', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': 'bcaff823-c5ca-4700-8c2e-811b4c792da9', 'content-type': 'text/xml', 'content-length': '224', 'date': 'Fri, 21 Jan 2022 10:59:10 GMT'}, 'RetryAttempts': 0}}

How to list all the SNS Topics tags using Boto3?

from typing import List

import boto3

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


def list_tags_for_sns_resource(topic_arn: str):
    """Lists tags for SNS Topic.

    :param topic_arn: ARN of SNS Topic to be listed tags.
    :return: JSON containing TopicARN.
    """
    response = sns_client.list_tags_for_resource(
        ResourceArn=topic_arn,
    )
    return response

Sample Call
list_tags_for_sns_resource("arn:aws:sns:us-east-1:123456789012:test_topic")


Sample Response
{'Tags': [{'Key': 'test_tag_key', 'Value': 'test_tag_value'}], 'ResponseMetadata': {'RequestId': '051e0c7f-7769-4f2d-88c5-0f53720c9121', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': '051e0c7f-7769-4f2d-88c5-0f53720c9121', 'content-type': 'text/xml', 'content-length': '402', 'date': 'Fri, 21 Jan 2022 10:59:10 GMT'}, 'RetryAttempts': 0}}

How to remove tags from SNS Topic using Boto3?

from typing import List

import boto3

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


def untag_sns_resource(topic_arn: str, tags: List[str]):
    """Untags SNS Topic.

    :param topic_arn: ARN of SNS Topic to be untagged.
    :param tags: Dictionary of tags to be removed from SNS Topic.
    :return: JSON containing TopicARN.
    """
    response = sns_client.untag_resource(
        ResourceArn=topic_arn,
        TagKeys=tags,
    )
    return response

Sample Call
untag_sns_resource("arn:aws:sns:us-east-1:123456789012:test_topic", tags=["test_tag_key"])

Sample Response
{'ResponseMetadata': {'RequestId': '9a563f26-54d6-42cf-9632-2b53d90a9ef8', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': '9a563f26-54d6-42cf-9632-2b53d90a9ef8', 'content-type': 'text/xml', 'content-length': '230', 'date': 'Fri, 21 Jan 2022 10:59:11 GMT'}, 'RetryAttempts': 0}}

How to publish messages to SNS Topic using Boto3?

from typing import List

import boto3

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


def publish_sns_message(topic_arn: str, message: str):
    """Publishes SNS Message.

    :param topic_arn: ARN of SNS Topic to be published to.
    :param message: Message to be published to SNS Topic.
    :return: JSON containing TopicARN.
    """
    response = sns_client.publish(
        TopicArn=topic_arn,
        Message=message,
    )
    return response

Sample Call
publish_sns_message("arn:aws:sns:us-east-1:123456789012:test_topic", message="test_message")


Sample Response
{'MessageId': 'd9e5b7ff-7f5e-4937-b5cc-a8462baa8d83', 'ResponseMetadata': {'RequestId': '822155fa-fb18-4c4d-89fa-0c8ed3dfc331', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': '822155fa-fb18-4c4d-89fa-0c8ed3dfc331', 'content-type': 'text/xml', 'content-length': '294', 'date': 'Fri, 21 Jan 2022 10:59:11 GMT'}, 'RetryAttempts': 0}}

How to publish batch messages to SNS Topic using Boto3?

from typing import List

import boto3

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


def publish_batch_sns_message(topic_arn: str, messages: List[dict]):
    """Publishes SNS Messages.

    :param topic_arn: ARN of SNS Topic to be published to.
    :param messages: List of messages to be published to SNS Topic.
    :return: JSON containing TopicARN.
    """
    response = sns_client.publish_batch(
        TopicArn=topic_arn,
        PublishBatchRequestEntries=messages,
    )
    return response

Sample Call

publish_batch_sns_message("arn:aws:sns:us-east-1:123456789012:test_topic", messages=[{"Id": "1234", "Message": "test_message"}])


Sample Response
{'Successful': [{'Id': '1234', 'MessageId': '12ffdf0c-7719-4baa-b2c7-4c8c812ca491'}], 'Failed': [], 'ResponseMetadata': {'RequestId': '3e00bc61-6d05-4988-a60a-3287f3756c82', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': '3e00bc61-6d05-4988-a60a-3287f3756c82', 'content-type': 'text/xml', 'content-length': '420', 'date': 'Fri, 21 Jan 2022 10:59:11 GMT'}, 'RetryAttempts': 0}}

How to subscribe to SNS Topic using Boto3?

from typing import List

import boto3

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


def subscribe_to_sns_topic(topic_arn: str, protocol: str, endpoint: str):
    """Subscribes to SNS Topic.

    :param topic_arn: ARN of SNS Topic to be subscribed to.
    :param protocol: Protocol to be subscribed to SNS Topic.
    :param endpoint: Endpoint to be subscribed to SNS Topic.
    :return: JSON containing TopicARN.
    """
    response = sns_client.subscribe(
        TopicArn=topic_arn,
        Protocol=protocol,
        Endpoint=endpoint,
    )
    return response

Sample Call

subscribe_to_sns_topic("arn:aws:sns:us-east-1:123456789012:test_topic", protocol="email", endpoint="test@gmail.com")

Sample Response
{'SubscriptionArn': 'pending confirmation', 'ResponseMetadata': {'RequestId': 'd2d753e9-59c6-47ea-a280-5960d796d3e9', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': 'd2d753e9-59c6-47ea-a280-5960d796d3e9', 'content-type': 'text/xml', 'content-length': '298', 'date': 'Fri, 21 Jan 2022 10:59:12 GMT'}, 'RetryAttempts': 0}}

How to set SNS subscription attributes using Boto3?

from typing import List

import boto3

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


def set_sns_subscription_attributes(subscription_arn: str, attribute_name: str, attribute_value: str):
    """Sets SNS Subscription Attributes.

    :param subscription_arn: ARN of SNS Subscription to be set attributes.
    :param attribute_name: Name of attribute to be set.
    :param attribute_value: Value of attribute to be set.
    :return: JSON containing TopicARN.
    """
    response = sns_client.set_subscription_attributes(
        SubscriptionArn=subscription_arn,
        AttributeName=attribute_name,
        AttributeValue=attribute_value,
    )
    return response

Sample Response
None

How to get SNS subscription attributes using Boto3?

from typing import List

import boto3

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


def get_sns_subscription_attributes(subscription_arn: str):
    """Gets SNS Subscription Attributes.

    :param subscription_arn: ARN of SNS Subscription to be get attributes.
    :return: JSON containing TopicARN.
    """
    response = sns_client.get_subscription_attributes(
        SubscriptionArn=subscription_arn,
    )
    return response

How to list all SNS subscriptions using Boto3?

from typing import List

import boto3

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


def list_sns_subscriptions():
    """Lists SNS Subscriptions."""
    response = sns_client.list_subscriptions()
    return response

Sample Call
list_sns_subscriptions()


Sample Response
{'Subscriptions': [{'SubscriptionArn': 'PendingConfirmation', 'Owner': '123456789012', 'Protocol': 'email', 'Endpoint': 'test@gmail.com', 'TopicArn': 'arn:aws:sns:us-east-1:123456789012:test_topic'}], 'NextToken': '83a28fcb16524fd2946cbd0009bb848f83a28fidopwncu==', 'ResponseMetadata': {'RequestId': '537822c8-23b9-4fb7-8a4d-65fbaa279def', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': '537822c8-23b9-4fb7-8a4d-65fbaa279def', 'content-type': 'text/xml', 'content-length': '681', 'date': 'Fri, 21 Jan 2022 10:59:12 GMT'}, 'RetryAttempts': 0}}

How to list all the SNS subscriptions for a particular topic using Boto3?

from typing import List

import boto3

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


def list_sns_subscriptions_by_topic(topic_arn: str):
    """Lists SNS Subscriptions by Topic.

    :param topic_arn: ARN of SNS Topic to be listed subscriptions.
    :return: JSON containing TopicARN.
    """
    response = sns_client.list_subscriptions_by_topic(
        TopicArn=topic_arn,
    )
    return response

Sample Call
list_sns_subscriptions_by_topic("arn:aws:sns:us-east-1:123456789012:test_topic")


Sample Response
{'Subscriptions': [{'SubscriptionArn': 'PendingConfirmation', 'Owner': '123456789012', 'Protocol': 'email', 'Endpoint': 'test@gmail.com', 'TopicArn': 'arn:aws:sns:us-east-1:123456789012:test_topic'}], 'ResponseMetadata': {'RequestId': 'aa9c5682-549d-4807-a720-495d226e4adf', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': 'aa9c5682-549d-4807-a720-495d226e4adf', 'content-type': 'text/xml', 'content-length': '633', 'date': 'Fri, 21 Jan 2022 10:59:12 GMT'}, 'RetryAttempts': 0}}

How to unsubscribe from the SNS topics using Boto3?

from typing import List

import boto3

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


def unsubscribe_sns(topic_arn: str, subscription_arn: str):
    """Unsubscribes SNS Topic.

    :param topic_arn: ARN of SNS Topic to be subscribed to.
    :param subscription_arn: ARN of SNS Subscription to be unsubscribed.
    :return: JSON containing TopicARN.
    """
    response = sns_client.unsubscribe(
        TopicArn=topic_arn,
        SubscriptionArn=subscription_arn,
    )
    return response

Sample Response
None

How to delete SNS Topic using Boto3?

from typing import List

import boto3

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


def delete_sns_topic(topic_arn: str):
    """Deletes SNS Topic.

    :param topic_arn: ARN of SNS Topic to be deleted.
    :return: JSON containing TopicARN.
    """
    response = sns_client.delete_topic(
        TopicArn=topic_arn,
    )
    return response

Sample Call
delete_sns_topic("arn:aws:sns:us-east-1:123456789012:test_topic")


Sample Response
{'ResponseMetadata': {'RequestId': 'ffefbd21-125a-4b2f-8879-6b977d821ccf', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': 'ffefbd21-125a-4b2f-8879-6b977d821ccf', 'content-type': 'text/xml', 'content-length': '201', 'date': 'Fri, 21 Jan 2022 10:59:13 GMT'}, 'RetryAttempts': 0}}

Conclusion

In this article, we discussed how to perform various SNS operations using Python and Boto3.

Related Articles

AWS SNS: The Complete Guide
AWS SNS: What it is, how does it work and what are the benefits?
AWS SNS: The Complete Guide to Python and BOTO3 Interface
AWS SNS: The Complete Guide to AWS-CLI SNS Commands