boto3_sns_topic_message_publish_quickstart.py

1import logging
2import boto3
3from botocore.exceptions import ClientError
4
5logger = logging.getLogger(__name__)
6
7def publish_message(topic_arn, message):
8    """
9    Publishes a message to a topic.
10
11    :param topic_arn: The ARN of the topic to publish to.
12    :param message: The message to publish.
13    :return: The ID of the message.
14    """
15    try:
16        sns_client = boto3.client("sns")
17        response = sns_client.publish(
18            TopicArn=topic_arn,
19            Message=message,
20        )
21        message_id = response["MessageId"]
22        logger.info("Published message to topic %s.", topic_arn)
23    except ClientError:
24        logger.exception("Couldn't publish message to topic %s.", topic_arn)
25        raise
26    else:
27        return message_id
28
29if __name__ == "__main__":
30    # Replace with your actual Topic ARN
31    TOPIC_ARN = "arn:aws:sns:us-west-2:123456789012:MyTopic"
32    MESSAGE = "Hello from the SNS Python Quickstart!"
33    
34    print(f"Publishing message to {TOPIC_ARN}...")
35    msg_id = publish_message(TOPIC_ARN, MESSAGE)
36    print(f"Message published with ID: {msg_id}")