sns_example.py

"""An AWS SNS example."""
import logging
import time

import boto3


logging.basicConfig(
    level=logging.INFO,
    format='[%(asctime)s] %(levelname)s:%(name)s:%(message)s'
)


class Topic:
    """An AWS SNS Topic."""

    def __init__(
        self,
        name,
        arn=None,
        logger=logging.getLogger('Topic'),
        client=boto3.client('sns')
    ):
        """Initialize topic."""
        self.name = name
        self.arn = arn
        self.logger = logger
        self.client = client

        if not self.arn:
            self.create()

    def create(self):
        """Create the topic if it doesn't exist."""
        response = self.client.create_topic(Name=self.name)
        self.arn = response['TopicArn']

        self.logger.info(response)

    def delete(self):
        """Delete the topic."""
        self.logger.info('Deleting %s', self.arn)
        self.client.delete_topic(TopicArn=self.arn)

    def __repr__(self):
        """Override the string representation method."""
        return 'Topic(name={}, arn={})'.format(self.name, self.arn)

    def __str__(self):
        """Override the string method."""
        return self.name


def produce(topic, logger=logging.getLogger('Producer'), count=1, interval=0):
    """Publish messages to a topic."""
    message = 'This is message number {} to topic: {}.'

    for i in range(0, count):
        if i > 0:
            time.sleep(interval)

        response = topic.client.publish(
            TopicArn=topic.arn,
            Message=message.format(i + 1, topic.name)
        )

        logger.info(response)


if __name__ == '__main__':
    produce(Topic('JimTestTopic'))