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.
Table of Contents
- How to create SNS Topic using Boto3?
- How to list all SNS Topics using Boto3?
- How to add permissions to existing SNS Topics using Boto3?
- How to remove SNS Topic permissions using Boto3?
- How to set SNS Topic attributes using Boto3?
- How to get SNS Topic attributes using Boto3?
- How to add tags to SNS Topic using Boto3?
- How to list all the SNS Topics tags using Boto3?
- How to remove tags from SNS Topic using Boto3?
- How to publish messages to SNS Topic using Boto3?
- How to publish batch messages to SNS Topic using Boto3?
- How to subscribe to SNS Topic using Boto3?
- How to set SNS subscription attributes using Boto3?
- How to get SNS subscription attributes using Boto3?
- How to list all SNS subscriptions using Boto3?
- How to list all the SNS subscriptions for a particular topic using Boto3?
- How to unsubscribe from the SNS topics using Boto3?
- How to delete SNS Topic using Boto3?
- Conclusion
- Related Articles
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}}
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}}
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}}
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
Recent Comments